This is an automated email from the ASF dual-hosted git repository.
wchevreuil pushed a commit to branch HBASE-30189
in repository https://gitbox.apache.org/repos/asf/hbase-connectors.git
The following commit(s) were added to refs/heads/HBASE-30189 by this push:
new 2c15325 Port pure utility classes with no Spark SQL dependency (#159)
2c15325 is described below
commit 2c15325ffe431cb393c58f6bad42859215e808fb
Author: Wellington Ramos Chevreuil <[email protected]>
AuthorDate: Wed Aug 26 12:02:09 2026 +0100
Port pure utility classes with no Spark SQL dependency (#159)
Co-authored-by: Claude Code (claude-opus-4-6) <[email protected]>
Signed-off-by: Peter Somogyi <[email protected]>
---
.../hbase/spark/datasources/NaiveEncoder.scala | 2 +-
.../spark/datasources/NaiveEncoderSuite.scala | 106 +++++++++++++++++++
.../hadoop/hbase/spark/HBaseConnectionCache.scala | 5 +-
.../hadoop/hbase/spark/datasources/Bound.scala | 116 +++++++++++++++++++++
.../spark/datasources/DataTypeParserWrapper.scala | 32 ++++++
.../hbase/spark/datasources/HBaseSparkConf.scala | 76 ++++++++++++++
.../hbase/spark/datasources/NaiveEncoder.scala | 51 +++------
.../datasources/SerializableConfiguration.scala | 45 ++++++++
.../hadoop/hbase/spark/datasources/package.scala | 45 ++++++++
.../spark/datasources/NaiveEncoderSuite.scala | 112 ++++++++++++++++++++
10 files changed, 551 insertions(+), 39 deletions(-)
diff --git
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoder.scala
b/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoder.scala
index b54d279..27adb17 100644
---
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoder.scala
+++
b/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoder.scala
@@ -243,7 +243,7 @@ class NaiveEncoder extends BytesEncoder with Logging {
Bytes.putDouble(result, 1, value.asInstanceOf[Double])
result
case BinaryType =>
- val v = value.asInstanceOf[Array[Bytes]]
+ val v = value.asInstanceOf[Array[Byte]]
val result = new Array[Byte](v.length + 1)
result(0) = BinaryEnc
System.arraycopy(v, 0, result, 1, v.length)
diff --git
a/spark/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoderSuite.scala
b/spark/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoderSuite.scala
new file mode 100644
index 0000000..374f09f
--- /dev/null
+++
b/spark/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoderSuite.scala
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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 org.apache.hadoop.hbase.spark.datasources
+
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.spark.sql.types._
+import org.scalatest.FunSuite
+
+class NaiveEncoderSuite extends FunSuite {
+
+ val encoder = new NaiveEncoder
+
+ test("encode BinaryType should preserve bytes") {
+ val input = Array[Byte](1, 2, 3, 4, 5)
+ val encoded = encoder.encode(BinaryType, input)
+ assert(encoded.length == input.length + 1)
+ assert(encoded(0) == encoder.BinaryEnc)
+ assert(encoded.slice(1, encoded.length).sameElements(input))
+ }
+
+ test("encode IntegerType") {
+ val encoded = encoder.encode(IntegerType, 42: Int)
+ assert(encoded.length == Bytes.SIZEOF_INT + 1)
+ assert(encoded(0) == encoder.IntEnc)
+ assert(Bytes.toInt(encoded, 1) == 42)
+ }
+
+ test("encode LongType") {
+ val encoded = encoder.encode(LongType, 123456789L)
+ assert(encoded.length == Bytes.SIZEOF_LONG + 1)
+ assert(encoded(0) == encoder.LongEnc)
+ assert(Bytes.toLong(encoded, 1) == 123456789L)
+ }
+
+ test("encode DoubleType") {
+ val encoded = encoder.encode(DoubleType, 3.14d)
+ assert(encoded.length == Bytes.SIZEOF_DOUBLE + 1)
+ assert(encoded(0) == encoder.DoubleEnc)
+ assert(Bytes.toDouble(encoded, 1) == 3.14d)
+ }
+
+ test("encode FloatType") {
+ val encoded = encoder.encode(FloatType, 2.5f)
+ assert(encoded.length == Bytes.SIZEOF_FLOAT + 1)
+ assert(encoded(0) == encoder.FloatEnc)
+ assert(Bytes.toFloat(encoded, 1) == 2.5f)
+ }
+
+ test("encode ShortType") {
+ val encoded = encoder.encode(ShortType, 7: Short)
+ assert(encoded.length == Bytes.SIZEOF_SHORT + 1)
+ assert(encoded(0) == encoder.ShortEnc)
+ assert(Bytes.toShort(encoded, 1) == 7)
+ }
+
+ test("encode StringType") {
+ val encoded = encoder.encode(StringType, "hello")
+ val expected = Bytes.toBytes("hello")
+ assert(encoded.length == expected.length + 1)
+ assert(encoded(0) == encoder.StringEnc)
+ assert(encoded.slice(1, encoded.length).sameElements(expected))
+ }
+
+ test("encode BooleanType true") {
+ val encoded = encoder.encode(BooleanType, true)
+ assert(encoded.length == Bytes.SIZEOF_BOOLEAN + 1)
+ assert(encoded(0) == encoder.BooleanEnc)
+ assert(encoded(1) == (-1: Byte))
+ }
+
+ test("encode BooleanType false") {
+ val encoded = encoder.encode(BooleanType, false)
+ assert(encoded.length == Bytes.SIZEOF_BOOLEAN + 1)
+ assert(encoded(0) == encoder.BooleanEnc)
+ assert(encoded(1) == (0: Byte))
+ }
+
+ test("filter with IntegerType GreaterThan") {
+ val encoded = encoder.encode(IntegerType, 10: Int)
+ val input = Bytes.toBytes(20: Int)
+ assert(
+ encoder.filter(input, 0, input.length, encoded, 0, encoded.length,
JavaBytesEncoder.Greater))
+ }
+
+ test("filter with IntegerType LessThan") {
+ val encoded = encoder.encode(IntegerType, 10: Int)
+ val input = Bytes.toBytes(5: Int)
+ assert(
+ encoder.filter(input, 0, input.length, encoded, 0, encoded.length,
JavaBytesEncoder.Less))
+ }
+}
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseConnectionCache.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseConnectionCache.scala
index 70b4c62..3f1cf3a 100644
---
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseConnectionCache.scala
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseConnectionCache.scala
@@ -29,6 +29,7 @@ import org.apache.hadoop.hbase.client.Table
import org.apache.hadoop.hbase.ipc.RpcControllerFactory
import org.apache.hadoop.hbase.security.User
import org.apache.hadoop.hbase.security.UserProvider
+import org.apache.hadoop.hbase.spark.datasources.HBaseSparkConf
import org.apache.yetus.audience.InterfaceAudience
import scala.collection.mutable
@@ -39,9 +40,7 @@ private[spark] object HBaseConnectionCache extends Logging {
val cacheStat = HBaseConnectionCacheStat(0, 0, 0)
- // in milliseconds
- private final val DEFAULT_TIME_OUT: Long = 10 * 60 * 1000
- private var timeout = DEFAULT_TIME_OUT
+ private var timeout: Long = HBaseSparkConf.DEFAULT_CONNECTION_CLOSE_DELAY
private var closed: Boolean = false
var housekeepingThread = new Thread(new Runnable {
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/Bound.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/Bound.scala
new file mode 100644
index 0000000..7b12982
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/Bound.scala
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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 org.apache.hadoop.hbase.spark.datasources
+
+import org.apache.yetus.audience.InterfaceAudience
+
+/**
+ * The Bound represent the boudary for the scan
+ *
+ * @param b The byte array of the bound
+ * @param inc inclusive or not.
+ */
[email protected]
+case class Bound(b: Array[Byte], inc: Boolean)
+
[email protected]
+case class Range(lower: Option[Bound], upper: Option[Bound])
+
[email protected]
+object Range {
+ def apply(region: HBaseRegion): Range = {
+ Range(
+ region.start.map(Bound(_, true)),
+ if (region.end.get.length == 0) {
+ None
+ } else {
+ region.end.map(Bound(_, false))
+ })
+ }
+}
+
[email protected]
+object Ranges {
+ def and(r: Range, rs: Seq[Range]): Seq[Range] = {
+ rs.flatMap { s =>
+ val lower = s.lower
+ .map { x =>
+ r.lower
+ .map { y =>
+ if (ord.compare(x.b, y.b) < 0) {
+ Some(y)
+ } else {
+ Some(x)
+ }
+ }
+ .getOrElse(Some(x))
+ }
+ .getOrElse(r.lower)
+
+ val upper = s.upper
+ .map { x =>
+ r.upper
+ .map { y =>
+ if (ord.compare(x.b, y.b) >= 0) {
+ Some(y)
+ } else {
+ Some(x)
+ }
+ }
+ .getOrElse(Some(x))
+ }
+ .getOrElse(r.upper)
+
+ val c = lower
+ .map { x =>
+ upper
+ .map { y =>
+ ord.compare(x.b, y.b)
+ }
+ .getOrElse(-1)
+ }
+ .getOrElse(-1)
+ if (c < 0) {
+ Some(Range(lower, upper))
+ } else {
+ None
+ }
+ }
+ }
+}
+
[email protected]
+object Points {
+ def and(r: Range, ps: Seq[Array[Byte]]): Seq[Array[Byte]] = {
+ ps.flatMap { p =>
+ if (ord.compare(r.lower.get.b, p) <= 0) {
+ if (r.upper.isDefined) {
+ if (ord.compare(r.upper.get.b, p) > 0) {
+ Some(p)
+ } else {
+ None
+ }
+ } else {
+ Some(p)
+ }
+ } else {
+ None
+ }
+ }
+ }
+}
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/DataTypeParserWrapper.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/DataTypeParserWrapper.scala
new file mode 100644
index 0000000..936276d
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/DataTypeParserWrapper.scala
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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 org.apache.hadoop.hbase.spark.datasources
+
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
+import org.apache.spark.sql.types.DataType
+import org.apache.yetus.audience.InterfaceAudience
+
[email protected]
+trait DataTypeParser {
+ def parse(dataTypeString: String): DataType
+}
+
[email protected]
+object DataTypeParserWrapper extends DataTypeParser {
+ def parse(dataTypeString: String): DataType =
CatalystSqlParser.parseDataType(dataTypeString)
+}
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseSparkConf.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseSparkConf.scala
new file mode 100644
index 0000000..b236af0
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseSparkConf.scala
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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 org.apache.hadoop.hbase.spark.datasources
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * This is the hbase configuration. User can either set them in SparkConf,
which
+ * will take effect globally, or configure it per table, which will overwrite
the value
+ * set in SparkConf. If not set, the default value will take effect.
+ */
[email protected]
+object HBaseSparkConf {
+
+ /**
+ * Set to false to disable server-side caching of blocks for this scan,
+ * false by default, since full table scans generate too much BC churn.
+ */
+ val QUERY_CACHEBLOCKS = "hbase.spark.query.cacheblocks"
+ val DEFAULT_QUERY_CACHEBLOCKS = false
+
+ /** The number of rows for caching that will be passed to scan. */
+ val QUERY_CACHEDROWS = "hbase.spark.query.cachedrows"
+
+ /** Set the maximum number of values to return for each call to next() in
scan. */
+ val QUERY_BATCHSIZE = "hbase.spark.query.batchsize"
+
+ /** The number of BulkGets send to HBase. */
+ val BULKGET_SIZE = "hbase.spark.bulkget.size"
+ val DEFAULT_BULKGET_SIZE = 1000
+
+ /** Set to specify the location of hbase configuration file. */
+ val HBASE_CONFIG_LOCATION = "hbase.spark.config.location"
+
+ /** Set to specify whether create or use latest cached HBaseContext */
+ val USE_HBASECONTEXT = "hbase.spark.use.hbasecontext"
+ val DEFAULT_USE_HBASECONTEXT = true
+
+ /** Pushdown the filter to data source engine to increase the performance of
queries. */
+ val PUSHDOWN_COLUMN_FILTER = "hbase.spark.pushdown.columnfilter"
+ val DEFAULT_PUSHDOWN_COLUMN_FILTER = true
+
+ /** Class name of the encoder, which encode data types from Spark to HBase
bytes. */
+ val QUERY_ENCODER = "hbase.spark.query.encoder"
+ val DEFAULT_QUERY_ENCODER = classOf[NaiveEncoder].getCanonicalName
+
+ /** The timestamp used to filter columns with a specific timestamp. */
+ val TIMESTAMP = "hbase.spark.query.timestamp"
+
+ /** The starting timestamp used to filter columns with a specific range of
versions. */
+ val TIMERANGE_START = "hbase.spark.query.timerange.start"
+
+ /** The ending timestamp used to filter columns with a specific range of
versions. */
+ val TIMERANGE_END = "hbase.spark.query.timerange.end"
+
+ /** The maximum number of version to return. */
+ val MAX_VERSIONS = "hbase.spark.query.maxVersions"
+
+ /** Delayed time to close hbase-spark connection when no reference to this
connection, in milliseconds. */
+ val DEFAULT_CONNECTION_CLOSE_DELAY = 10 * 60 * 1000
+}
\ No newline at end of file
diff --git
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoder.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoder.scala
similarity index 86%
copy from
spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoder.scala
copy to
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoder.scala
index b54d279..2e32fae 100644
---
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoder.scala
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoder.scala
@@ -16,26 +16,9 @@
* limitations under the License.
*/
package org.apache.hadoop.hbase.spark.datasources
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file 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.
- */
import org.apache.hadoop.hbase.spark.Logging
import
org.apache.hadoop.hbase.spark.datasources.JavaBytesEncoder.JavaBytesEncoder
-import org.apache.hadoop.hbase.spark.hbase._
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
@@ -55,16 +38,16 @@ class NaiveEncoder extends BytesEncoder with Logging {
code += 1
(code - 1).asInstanceOf[Byte]
}
- val BooleanEnc = nextCode
- val ShortEnc = nextCode
- val IntEnc = nextCode
- val LongEnc = nextCode
- val FloatEnc = nextCode
- val DoubleEnc = nextCode
- val StringEnc = nextCode
- val BinaryEnc = nextCode
- val TimestampEnc = nextCode
- val UnknownEnc = nextCode
+ val BooleanEnc: Byte = nextCode
+ val ShortEnc: Byte = nextCode
+ val IntEnc: Byte = nextCode
+ val LongEnc: Byte = nextCode
+ val FloatEnc: Byte = nextCode
+ val DoubleEnc: Byte = nextCode
+ val StringEnc: Byte = nextCode
+ val BinaryEnc: Byte = nextCode
+ val TimestampEnc: Byte = nextCode
+ val UnknownEnc: Byte = nextCode
/**
* Evaluate the java primitive type and return the BoundRanges. For one
value, it may have
@@ -79,7 +62,7 @@ class NaiveEncoder extends BytesEncoder with Logging {
* But the order of negative number is the reverse order of byte array.
Please refer to IEEE-754
* and https://en.wikipedia.org/wiki/Single-precision_floating-point_format
*/
- def ranges(in: Any): Option[BoundRanges] = in match {
+ override def ranges(in: Any): Option[BoundRanges] = in match {
case a: Integer =>
val b = Bytes.toBytes(a)
if (a >= 0) {
@@ -203,8 +186,8 @@ class NaiveEncoder extends BytesEncoder with Logging {
* encode the data type into byte array. Note that it is a naive
implementation with the
* data type byte appending to the head of the serialized byte array.
*
- * @param dt: The data type of the input
- * @param value: the value of the input
+ * @param dt : The data type of the input
+ * @param value : the value of the input
* @return the byte array with the first byte indicating the data type.
*/
override def encode(dt: DataType, value: Any): Array[Byte] = {
@@ -213,8 +196,8 @@ class NaiveEncoder extends BytesEncoder with Logging {
val result = new Array[Byte](Bytes.SIZEOF_BOOLEAN + 1)
result(0) = BooleanEnc
value.asInstanceOf[Boolean] match {
- case true => result(1) = -1: Byte
- case false => result(1) = 0: Byte
+ case true => result(1) = (-1: Byte)
+ case false => result(1) = (0: Byte)
}
result
case ShortType =>
@@ -243,7 +226,7 @@ class NaiveEncoder extends BytesEncoder with Logging {
Bytes.putDouble(result, 1, value.asInstanceOf[Double])
result
case BinaryType =>
- val v = value.asInstanceOf[Array[Bytes]]
+ val v = value.asInstanceOf[Array[Byte]]
val result = new Array[Byte](v.length + 1)
result(0) = BinaryEnc
System.arraycopy(v, 0, result, 1, v.length)
@@ -293,8 +276,6 @@ class NaiveEncoder extends BytesEncoder with Logging {
val value = Bytes.toDouble(filterBytes, offset2 + 1)
compare(in.compareTo(value), ops)
case _ =>
- // for String, Byte, Binary, Boolean and other types
- // we can use the order of byte array directly.
compare(
Bytes.compareTo(input, offset1, length1, filterBytes, offset2 + 1,
length2 - 1),
ops)
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/SerializableConfiguration.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/SerializableConfiguration.scala
new file mode 100644
index 0000000..552abda
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/SerializableConfiguration.scala
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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 org.apache.hadoop.hbase.spark.datasources
+
+import java.io.{IOException, ObjectInputStream, ObjectOutputStream}
+import org.apache.hadoop.conf.Configuration
+import org.apache.yetus.audience.InterfaceAudience
+import scala.util.control.NonFatal
+
[email protected]
+class SerializableConfiguration(@transient var value: Configuration) extends
Serializable {
+ private def writeObject(out: ObjectOutputStream): Unit = tryOrIOException {
+ out.defaultWriteObject()
+ value.write(out)
+ }
+
+ private def readObject(in: ObjectInputStream): Unit = tryOrIOException {
+ value = new Configuration(false)
+ value.readFields(in)
+ }
+
+ private def tryOrIOException(block: => Unit): Unit = {
+ try {
+ block
+ } catch {
+ case e: IOException => throw e
+ case NonFatal(t) => throw new IOException(t)
+ }
+ }
+}
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/package.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/package.scala
new file mode 100644
index 0000000..2292422
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/package.scala
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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 org.apache.hadoop.hbase.spark
+
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.spark.Partition
+import org.apache.yetus.audience.InterfaceAudience
+import scala.math.Ordering
+
+package object datasources {
+ type HBaseType = Array[Byte]
+ def bytesMin: Array[Byte] = new Array[Byte](0)
+ def bytesMax: Array[Byte] = null
+ val ByteMax: Byte = -1.asInstanceOf[Byte]
+ val ByteMin: Byte = 0.asInstanceOf[Byte]
+ val ord: Ordering[HBaseType] = new Ordering[HBaseType] {
+ def compare(x: Array[Byte], y: Array[Byte]): Int = {
+ Bytes.compareTo(x, y)
+ }
+ }
+ implicit val order: Ordering[HBaseType] = ord
+
+ @InterfaceAudience.Private
+ case class HBaseRegion(
+ override val index: Int,
+ start: Option[HBaseType] = None,
+ end: Option[HBaseType] = None,
+ server: Option[String] = None)
+ extends Partition
+}
diff --git
a/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoderSuite.scala
b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoderSuite.scala
new file mode 100644
index 0000000..2828f17
--- /dev/null
+++
b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/NaiveEncoderSuite.scala
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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 org.apache.hadoop.hbase.spark.datasources
+
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.spark.sql.types._
+import org.scalatest.funsuite.AnyFunSuite
+
+class NaiveEncoderSuite extends AnyFunSuite {
+
+ val encoder = new NaiveEncoder
+
+ test("encode BinaryType should preserve bytes") {
+ val input = Array[Byte](1, 2, 3, 4, 5)
+ val encoded = encoder.encode(BinaryType, input)
+ assert(encoded.length == input.length + 1)
+ assert(encoded(0) == encoder.BinaryEnc)
+ assert(encoded.slice(1, encoded.length).sameElements(input))
+ }
+
+ test("encode IntegerType") {
+ val encoded = encoder.encode(IntegerType, 42: Int)
+ assert(encoded.length == Bytes.SIZEOF_INT + 1)
+ assert(encoded(0) == encoder.IntEnc)
+ assert(Bytes.toInt(encoded, 1) == 42)
+ }
+
+ test("encode LongType") {
+ val encoded = encoder.encode(LongType, 123456789L)
+ assert(encoded.length == Bytes.SIZEOF_LONG + 1)
+ assert(encoded(0) == encoder.LongEnc)
+ assert(Bytes.toLong(encoded, 1) == 123456789L)
+ }
+
+ test("encode DoubleType") {
+ val encoded = encoder.encode(DoubleType, 3.14d)
+ assert(encoded.length == Bytes.SIZEOF_DOUBLE + 1)
+ assert(encoded(0) == encoder.DoubleEnc)
+ assert(Bytes.toDouble(encoded, 1) == 3.14d)
+ }
+
+ test("encode FloatType") {
+ val encoded = encoder.encode(FloatType, 2.5f)
+ assert(encoded.length == Bytes.SIZEOF_FLOAT + 1)
+ assert(encoded(0) == encoder.FloatEnc)
+ assert(Bytes.toFloat(encoded, 1) == 2.5f)
+ }
+
+ test("encode ShortType") {
+ val encoded = encoder.encode(ShortType, 7: Short)
+ assert(encoded.length == Bytes.SIZEOF_SHORT + 1)
+ assert(encoded(0) == encoder.ShortEnc)
+ assert(Bytes.toShort(encoded, 1) == 7)
+ }
+
+ test("encode StringType") {
+ val encoded = encoder.encode(StringType, "hello")
+ val expected = Bytes.toBytes("hello")
+ assert(encoded.length == expected.length + 1)
+ assert(encoded(0) == encoder.StringEnc)
+ assert(encoded.slice(1, encoded.length).sameElements(expected))
+ }
+
+ test("encode BooleanType true") {
+ val encoded = encoder.encode(BooleanType, true)
+ assert(encoded.length == Bytes.SIZEOF_BOOLEAN + 1)
+ assert(encoded(0) == encoder.BooleanEnc)
+ assert(encoded(1) == (-1: Byte))
+ }
+
+ test("encode BooleanType false") {
+ val encoded = encoder.encode(BooleanType, false)
+ assert(encoded.length == Bytes.SIZEOF_BOOLEAN + 1)
+ assert(encoded(0) == encoder.BooleanEnc)
+ assert(encoded(1) == (0: Byte))
+ }
+
+ test("filter with IntegerType GreaterThan") {
+ val encoded = encoder.encode(IntegerType, 10: Int)
+ val input = Bytes.toBytes(20: Int)
+ assert(
+ encoder.filter(
+ input, 0, input.length,
+ encoded, 0, encoded.length,
+ JavaBytesEncoder.Greater))
+ }
+
+ test("filter with IntegerType LessThan") {
+ val encoded = encoder.encode(IntegerType, 10: Int)
+ val input = Bytes.toBytes(5: Int)
+ assert(
+ encoder.filter(
+ input, 0, input.length,
+ encoded, 0, encoded.length,
+ JavaBytesEncoder.Less))
+ }
+}
