This is an automated email from the ASF dual-hosted git repository.
wchevreuil pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hbase-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new 267b6dc HBASE-30184 Move SparkSQLPushDownFilter and related classes
to a common module (#156)
267b6dc is described below
commit 267b6dc9bfe6e7a0e42ed7411c976dd7019f8576
Author: Wellington Ramos Chevreuil <[email protected]>
AuthorDate: Fri Jun 26 14:34:51 2026 +0100
HBASE-30184 Move SparkSQLPushDownFilter and related classes to a common
module (#156)
Signed-off-by: Tak Lon (Stephen) Wu <[email protected]>
Co-authored-by: Cursor Composer 2.5 <[email protected]>
---
hbase-connectors-assembly/pom.xml | 16 +++++++++
.../src/main/assembly/hbase-connectors-bin.xml | 9 +++--
pom.xml | 2 ++
.../hadoop/hbase/spark/PushdownMappedField.java} | 36 +++++++------------
.../hadoop/hbase/spark/SparkSQLPushDownFilter.java | 12 ++++---
.../hadoop/hbase/spark/ByteArrayComparable.scala | 0
.../hbase/spark/DynamicLogicExpression.scala | 0
.../hbase/spark/datasources/JavaBytesEncoder.scala | 41 +++++++++++++++-------
.../apache/hadoop/hbase/spark/DefaultSource.scala | 12 ++++++-
9 files changed, 80 insertions(+), 48 deletions(-)
diff --git a/hbase-connectors-assembly/pom.xml
b/hbase-connectors-assembly/pom.xml
index 4bed23a..11016f5 100755
--- a/hbase-connectors-assembly/pom.xml
+++ b/hbase-connectors-assembly/pom.xml
@@ -234,4 +234,20 @@
<!-- /end building aggregation of NOTICE files -->
</plugins>
</build>
+
+ <profiles>
+ <!-- With -Pspark4, force the reactor to build spark4 before this module
(Maven otherwise
+ schedules assembly ahead of the spark4 aggregator; see
hbase-connectors-bin.xml). -->
+ <profile>
+ <id>spark4</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hbase.connectors.spark</groupId>
+ <artifactId>hbase-spark4</artifactId>
+ <version>${revision}</version>
+ <optional>true</optional>
+ </dependency>
+ </dependencies>
+ </profile>
+ </profiles>
</project>
diff --git
a/hbase-connectors-assembly/src/main/assembly/hbase-connectors-bin.xml
b/hbase-connectors-assembly/src/main/assembly/hbase-connectors-bin.xml
index f75ab36..07613c9 100755
--- a/hbase-connectors-assembly/src/main/assembly/hbase-connectors-bin.xml
+++ b/hbase-connectors-assembly/src/main/assembly/hbase-connectors-bin.xml
@@ -34,12 +34,11 @@
<useAllReactorProjects>true</useAllReactorProjects>
<excludes>
<exclude>org.apache.hbase.connectors.spark:hbase-spark-it</exclude>
- <!-- Not yet packaged when assembly runs (reactor lists spark4 after
this module).
- Exclude until dist layout includes Spark 4; -->
+ <!-- Spark 4 line: not part of the default binary layout yet. -->
<exclude>org.apache.hbase.connectors.spark:hbase-spark4</exclude>
- <!-- Keep the Scala binary suffix in sync with scala.binary.version in
spark4/pom.xml;
- hbase-spark-pushdown artifactId is version-suffixed. -->
-
<exclude>org.apache.hbase.connectors.spark:hbase-spark-pushdown_${scala.binary.version}</exclude>
+ <!-- Internal pushdown libraries (pulled transitively by hbase-spark /
hbase-spark4).
+ ArtifactIds are Scala-suffixed; assembly descriptors do not
interpolate ${scala.binary.version}. -->
+
<exclude>org.apache.hbase.connectors.spark:hbase-spark_${spark4.scala.binary.version}</exclude>
</excludes>
<binaries>
<unpack>false</unpack>
diff --git a/pom.xml b/pom.xml
index 5257ab6..0bf8840 100644
--- a/pom.xml
+++ b/pom.xml
@@ -213,6 +213,8 @@
-Djdk.reflect.useDirectMethodHandle=false
-Dio.netty.tryReflectionSetAccessible=true</spark-surefire.jdk11.flags>
<spark-surefire.jdk17.flags/>
+ <!-- spark4/hbase-spark-pushdown artifactId suffix; exclude is a no-op
unless -Pspark4 -->
+ <spark4.scala.binary.version>2.13</spark4.scala.binary.version>
</properties>
<dependencyManagement>
<dependencies>
diff --git
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/ByteArrayComparable.scala
b/spark/hbase-spark-pushdown/src/main/java/org/apache/hadoop/hbase/spark/PushdownMappedField.java
similarity index 51%
copy from
spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/ByteArrayComparable.scala
copy to
spark/hbase-spark-pushdown/src/main/java/org/apache/hadoop/hbase/spark/PushdownMappedField.java
index 78cd3ab..1ef316a 100644
---
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/ByteArrayComparable.scala
+++
b/spark/hbase-spark-pushdown/src/main/java/org/apache/hadoop/hbase/spark/PushdownMappedField.java
@@ -15,33 +15,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.hadoop.hbase.spark
+package org.apache.hadoop.hbase.spark;
-import org.apache.hadoop.hbase.util.Bytes
-import org.apache.yetus.audience.InterfaceAudience
+import org.apache.yetus.audience.InterfaceAudience;
[email protected]
-class ByteArrayComparable(val bytes: Array[Byte], val offset: Int = 0, var
length: Int = -1)
- extends Comparable[ByteArrayComparable] {
-
- if (length == -1) {
- length = bytes.length
- }
+/**
+ * Column mapping for {@link SparkSQLPushDownFilter} built from datasource
catalog rows. Defined in
+ * Java so Spark 3 and Spark 4 connector modules can contribute mappings
without tying the filter to
+ * a single Scala {@code Field} type.
+ */
[email protected]
+public interface PushdownMappedField {
- override def compareTo(o: ByteArrayComparable): Int = {
- Bytes.compareTo(bytes, offset, length, o.bytes, o.offset, o.length)
- }
+ String colName();
- override def hashCode(): Int = {
- Bytes.hashCode(bytes, offset, length)
- }
+ byte[] cfBytes();
- override def equals(obj: Any): Boolean = {
- obj match {
- case b: ByteArrayComparable =>
- Bytes.equals(bytes, offset, length, b.bytes, b.offset, b.length)
- case _ =>
- false
- }
- }
+ byte[] colBytes();
}
diff --git
a/spark/hbase-spark/src/main/java/org/apache/hadoop/hbase/spark/SparkSQLPushDownFilter.java
b/spark/hbase-spark-pushdown/src/main/java/org/apache/hadoop/hbase/spark/SparkSQLPushDownFilter.java
similarity index 96%
rename from
spark/hbase-spark/src/main/java/org/apache/hadoop/hbase/spark/SparkSQLPushDownFilter.java
rename to
spark/hbase-spark-pushdown/src/main/java/org/apache/hadoop/hbase/spark/SparkSQLPushDownFilter.java
index 02034c1..a15b209 100644
---
a/spark/hbase-spark/src/main/java/org/apache/hadoop/hbase/spark/SparkSQLPushDownFilter.java
+++
b/spark/hbase-spark-pushdown/src/main/java/org/apache/hadoop/hbase/spark/SparkSQLPushDownFilter.java
@@ -28,14 +28,12 @@ import
org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.filter.Filter.ReturnCode;
import org.apache.hadoop.hbase.filter.FilterBase;
import org.apache.hadoop.hbase.spark.datasources.BytesEncoder;
-import org.apache.hadoop.hbase.spark.datasources.Field;
import org.apache.hadoop.hbase.spark.datasources.JavaBytesEncoder;
import org.apache.hadoop.hbase.spark.protobuf.generated.SparkFilterProtos;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import scala.collection.mutable.MutableList;
import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
import
org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
@@ -71,8 +69,12 @@ public class SparkSQLPushDownFilter extends FilterBase {
this.encoderClassName = encoderClassName;
}
+ /**
+ * @param columns Mapped columns projected for this scan/get (typically from
the catalog parser's
+ * {@code Field} values in either connector line).
+ */
public SparkSQLPushDownFilter(DynamicLogicExpression dynamicLogicExpression,
- byte[][] valueFromQueryArray, MutableList<Field> fields, String
encoderClassName) {
+ byte[][] valueFromQueryArray, List<PushdownMappedField> columns, String
encoderClassName) {
this.dynamicLogicExpression = dynamicLogicExpression;
this.valueFromQueryArray = valueFromQueryArray;
this.encoderClassName = encoderClassName;
@@ -80,8 +82,8 @@ public class SparkSQLPushDownFilter extends FilterBase {
// generate family qualifier to index mapping
this.currentCellToColumnIndexMap = new HashMap<>();
- for (int i = 0; i < fields.size(); i++) {
- Field field = fields.apply(i);
+ for (int i = 0; i < columns.size(); i++) {
+ PushdownMappedField field = columns.get(i);
byte[] cfBytes = field.cfBytes();
ByteArrayComparable familyByteComparable =
diff --git
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/ByteArrayComparable.scala
b/spark/hbase-spark-pushdown/src/main/scala/org/apache/hadoop/hbase/spark/ByteArrayComparable.scala
similarity index 100%
rename from
spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/ByteArrayComparable.scala
rename to
spark/hbase-spark-pushdown/src/main/scala/org/apache/hadoop/hbase/spark/ByteArrayComparable.scala
diff --git
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DynamicLogicExpression.scala
b/spark/hbase-spark-pushdown/src/main/scala/org/apache/hadoop/hbase/spark/DynamicLogicExpression.scala
similarity index 100%
rename from
spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DynamicLogicExpression.scala
rename to
spark/hbase-spark-pushdown/src/main/scala/org/apache/hadoop/hbase/spark/DynamicLogicExpression.scala
diff --git
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/JavaBytesEncoder.scala
b/spark/hbase-spark-pushdown/src/main/scala/org/apache/hadoop/hbase/spark/datasources/JavaBytesEncoder.scala
similarity index 79%
rename from
spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/JavaBytesEncoder.scala
rename to
spark/hbase-spark-pushdown/src/main/scala/org/apache/hadoop/hbase/spark/datasources/JavaBytesEncoder.scala
index eac4feb..0789ed6 100644
---
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/JavaBytesEncoder.scala
+++
b/spark/hbase-spark-pushdown/src/main/scala/org/apache/hadoop/hbase/spark/datasources/JavaBytesEncoder.scala
@@ -18,11 +18,11 @@
package org.apache.hadoop.hbase.spark.datasources
import org.apache.hadoop.hbase.HBaseInterfaceAudience
-import org.apache.hadoop.hbase.spark.Logging
import
org.apache.hadoop.hbase.spark.datasources.JavaBytesEncoder.JavaBytesEncoder
import org.apache.spark.sql.types._
-import org.apache.yetus.audience.InterfaceAudience
-import org.apache.yetus.audience.InterfaceStability
+import org.apache.yetus.audience.{InterfaceAudience, InterfaceStability}
+import org.slf4j.{Logger, LoggerFactory}
+import scala.util.Try
/**
* The ranges for the data type whose size is known. Whether the bound is
inclusive
@@ -98,23 +98,38 @@ trait BytesEncoder {
@InterfaceAudience.LimitedPrivate(Array(HBaseInterfaceAudience.SPARK))
@InterfaceStability.Evolving
-object JavaBytesEncoder extends Enumeration with Logging {
+object JavaBytesEncoder extends Enumeration {
type JavaBytesEncoder = Value
val Greater, GreaterEqual, Less, LessEqual, Equal, Unknown = Value
+ private val log: Logger =
+
LoggerFactory.getLogger("org.apache.hadoop.hbase.spark.datasources.JavaBytesEncoder")
+
+ private def naiveFallbackEncoder(): BytesEncoder = {
+ val candidates = Seq(
+ "org.apache.hadoop.hbase.spark4.datasources.NaiveEncoder",
+ "org.apache.hadoop.hbase.spark.datasources.NaiveEncoder")
+ candidates
+ .map(n =>
Try(Class.forName(n).newInstance.asInstanceOf[BytesEncoder]).toOption)
+ .flatten
+ .headOption
+ .getOrElse(throw new IllegalStateException(
+ "Neither Spark 3 nor Spark 4 NaiveEncoder was loadable while falling
back BytesEncoder"))
+ }
+
/**
* create the encoder/decoder
*
* @param clsName: the class name of the encoder/decoder class
* @return the instance of the encoder plugin.
*/
- def create(clsName: String): BytesEncoder = {
- try {
- Class.forName(clsName).newInstance.asInstanceOf[BytesEncoder]
- } catch {
- case _: Throwable =>
- logWarning(s"$clsName cannot be initiated, falling back to naive
encoder")
- new NaiveEncoder()
- }
- }
+ def create(clsName: String): BytesEncoder =
+ Try(Class.forName(clsName).newInstance.asInstanceOf[BytesEncoder]).recover
{
+ case t: Throwable =>
+ log.warn(
+ "{} cannot be initiated, falling back to naive encoder: {}",
+ clsName: Any,
+ String.valueOf(t): Any)
+ naiveFallbackEncoder()
+ }.get
}
diff --git
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
b/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
index 4ad945a..3d686dc 100644
---
a/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
+++
b/spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
@@ -36,6 +36,7 @@ import org.apache.spark.sql.{DataFrame, Row, SaveMode,
SQLContext}
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types._
import org.apache.yetus.audience.InterfaceAudience
+import scala.collection.JavaConverters._
import scala.collection.mutable
/**
@@ -391,11 +392,20 @@ case class HBaseRelation(
val pushDownFilterJava =
if (usePushDownColumnFilter && pushDownDynamicLogicExpression != null) {
+ val columnMappings =
+ requiredQualifierDefinitionList.map {
+ field =>
+ new PushdownMappedField {
+ override def colName(): String = field.colName
+ override def cfBytes(): Array[Byte] = field.cfBytes
+ override def colBytes(): Array[Byte] = field.colBytes
+ }
+ }
Some(
new SparkSQLPushDownFilter(
pushDownDynamicLogicExpression,
valueArray,
- requiredQualifierDefinitionList,
+ columnMappings.asJava,
encoderClsName))
} else {
None
