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 0c799b8 HBASE-30185 Port hbase-spark lower level RDD API to spark4
module (#157)
0c799b8 is described below
commit 0c799b8a9fc4133243035b083ae52631e9f72027
Author: Wellington Ramos Chevreuil <[email protected]>
AuthorDate: Mon Jul 13 11:52:00 2026 +0100
HBASE-30185 Port hbase-spark lower level RDD API to spark4 module (#157)
Co-authored-by: Claude Code (claude-opus-4-6) <[email protected]>
Signed-off-by: Peter Somogyi <[email protected]>
---
dev-support/jenkins/Dockerfile | 6 +
dev-support/jenkins/Jenkinsfile | 2 +-
dev-support/jenkins/hbase-personality.sh | 11 +
.../jenkins/jenkins_precommit_github_yetus.sh | 3 +-
pom.xml | 14 +-
spark4/hbase-spark-pushdown/pom.xml | 380 ++++++++++-----------
spark4/hbase-spark4/pom.xml | 99 +++++-
.../apache/hadoop/hbase/spark/HBaseContext.scala | 170 +++++++++
.../org/apache/hadoop/hbase/spark/Logging.scala | 101 ++++++
.../{Spark4Placeholder.scala => NewHBaseRDD.scala} | 25 +-
.../hadoop/hbase/spark/HBaseContextSuite.scala | 161 +++++++++
spark4/pom.xml | 14 +-
12 files changed, 774 insertions(+), 212 deletions(-)
diff --git a/dev-support/jenkins/Dockerfile b/dev-support/jenkins/Dockerfile
index ad68df9..45551b9 100644
--- a/dev-support/jenkins/Dockerfile
+++ b/dev-support/jenkins/Dockerfile
@@ -28,6 +28,12 @@ RUN apt-get -q update && apt-get -q install
--no-install-recommends -y \
apt-get clean && \
rm -rf /var/lib/apt/lists/*
+# JDK 17 for the Spark4 profile (maven-enforcer-plugin requires JDK 17+)
+RUN wget -q
https://github.com/adoptium/temurin17-binaries/releases/download/jdk-17.0.11%2B9/OpenJDK17U-jdk_x64_linux_hotspot_17.0.11_9.tar.gz
-O /tmp/jdk17.tar.gz && \
+ mkdir -p /usr/local/openjdk-17 && \
+ tar -xzf /tmp/jdk17.tar.gz --strip-components=1 -C /usr/local/openjdk-17
&& \
+ rm /tmp/jdk17.tar.gz
+
###
# Avoid out of memory errors in builds
###
diff --git a/dev-support/jenkins/Jenkinsfile b/dev-support/jenkins/Jenkinsfile
index eabf47b..d788de9 100644
--- a/dev-support/jenkins/Jenkinsfile
+++ b/dev-support/jenkins/Jenkinsfile
@@ -35,7 +35,7 @@ pipeline {
PATCH_REL = 'output'
YETUS_REL = 'yetus'
// Branch or tag name. Yetus release tags are 'rel/X.Y.Z'
- YETUS_VERSION = 'rel/0.12.0'
+ YETUS_VERSION = 'rel/0.15.0'
DOCKERFILE_REL = "${SRC_REL}/dev-support/jenkins/Dockerfile"
YETUS_DRIVER_REL =
"${SRC_REL}/dev-support/jenkins/jenkins_precommit_github_yetus.sh"
ARCHIVE_PATTERN_LIST =
'TEST-*.xml,org.apache.h*.txt,*.dumpstream,*.dump'
diff --git a/dev-support/jenkins/hbase-personality.sh
b/dev-support/jenkins/hbase-personality.sh
index 9ec5f3f..e3d899d 100755
--- a/dev-support/jenkins/hbase-personality.sh
+++ b/dev-support/jenkins/hbase-personality.sh
@@ -30,6 +30,17 @@ function personality_modules
local extra=""
local MODULES=("${CHANGED_MODULES[@]}")
+ # When spark4 modules are in the changed set, activate the spark4 profile
+ # so the full reactor (including hbase-spark-pushdown_2.13) is built.
+ # The maven-enforcer-plugin in the spark4 profile requires JDK 17+.
+ if [[ "${MODULES[*]}" =~ spark4 ]]; then
+ extra="${extra} -Pspark4"
+ if [[ "${JAVA_HOME}" != "/usr/local/openjdk-17" ]]; then
+ export JAVA_HOME=/usr/local/openjdk-17
+ export PATH="${JAVA_HOME}/bin:${PATH}"
+ fi
+ fi
+
clear_personality_queue
# Always to install at root.
diff --git a/dev-support/jenkins/jenkins_precommit_github_yetus.sh
b/dev-support/jenkins/jenkins_precommit_github_yetus.sh
index 5a03ada..5e2a2aa 100755
--- a/dev-support/jenkins/jenkins_precommit_github_yetus.sh
+++ b/dev-support/jenkins/jenkins_precommit_github_yetus.sh
@@ -85,8 +85,7 @@ YETUS_ARGS+=("--brief-report-file=${PATCHDIR}/brief.txt")
YETUS_ARGS+=("--console-report-file=${PATCHDIR}/console.txt")
YETUS_ARGS+=("--html-report-file=${PATCHDIR}/report.html")
# enable writing back to Github
-YETUS_ARGS+=("--github-password=${GITHUB_PASSWORD}")
-YETUS_ARGS+=("--github-user=${GITHUB_USER}")
+YETUS_ARGS+=("--github-token=${GITHUB_PASSWORD}")
# auto-kill any surefire stragglers during unit test runs
YETUS_ARGS+=("--reapermode=kill")
# set relatively high limits for ASF machines
diff --git a/pom.xml b/pom.xml
index 0bf8840..16d19ec 100644
--- a/pom.xml
+++ b/pom.xml
@@ -153,7 +153,7 @@
<log4j2.version>2.25.3</log4j2.version>
<reload4j.version>1.2.25</reload4j.version>
<checkstyle.version>8.45.1</checkstyle.version>
- <spotless.version>2.27.2</spotless.version>
+ <spotless.version>2.30.0</spotless.version>
<maven.checkstyle.version>3.2.1</maven.checkstyle.version>
<hbase.checkstyle.version>2.5.0</hbase.checkstyle.version>
<surefire.version>3.5.2</surefire.version>
@@ -215,6 +215,9 @@
<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>
+ <spark4.scalatest.version>3.2.17</spark4.scalatest.version>
+
<spark4.scalatest.maven.plugin.version>2.2.0</spark4.scalatest.maven.plugin.version>
+ <nashorn.version>15.4</nashorn.version>
</properties>
<dependencyManagement>
<dependencies>
@@ -809,6 +812,8 @@
<excludes>
<exclude>**/target/**</exclude>
<exclude>**/dependency-reduced-pom.xml</exclude>
+ <exclude>**/.flattened-pom.xml</exclude>
+ <exclude>.flattened-pom.xml</exclude>
</excludes>
<!-- define the steps to apply to those files -->
<trimTrailingWhitespace/>
@@ -839,6 +844,13 @@
</format>
</formats>
</configuration>
+ <dependencies>
+ <dependency>
+ <groupId>org.openjdk.nashorn</groupId>
+ <artifactId>nashorn-core</artifactId>
+ <version>${nashorn.version}</version>
+ </dependency>
+ </dependencies>
</plugin>
</plugins>
<extensions>
diff --git a/spark4/hbase-spark-pushdown/pom.xml
b/spark4/hbase-spark-pushdown/pom.xml
index 5dd98b0..eff501f 100644
--- a/spark4/hbase-spark-pushdown/pom.xml
+++ b/spark4/hbase-spark-pushdown/pom.xml
@@ -19,200 +19,198 @@
-->
<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
- <modelVersion>4.0.0</modelVersion>
+ <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.hbase.connectors</groupId>
- <artifactId>spark4</artifactId>
- <version>${revision}</version>
- <relativePath>../pom.xml</relativePath>
- </parent>
+ <parent>
+ <groupId>org.apache.hbase.connectors</groupId>
+ <artifactId>spark4</artifactId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
- <groupId>org.apache.hbase.connectors.spark</groupId>
- <artifactId>hbase-spark-pushdown_${scala.binary.version}</artifactId>
- <name>Apache HBase - Spark Connector SQL Pushdown (shared for Spark
4)</name>
- <description>
- Builds the same pushdown sources as spark/hbase-spark-pushdown against
Spark 4 / Scala 2.13 (see module sources).
- </description>
+ <groupId>org.apache.hbase.connectors.spark</groupId>
+ <artifactId>hbase-spark-pushdown_${scala.binary.version}</artifactId>
+ <name>Apache HBase - Spark Connector SQL Pushdown (shared for Spark 4)</name>
+ <description>Builds the same pushdown sources as spark/hbase-spark-pushdown
against Spark 4 / Scala 2.13 (see module sources).</description>
- <dependencies>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.hbase.thirdparty</groupId>
- <artifactId>hbase-shaded-miscellaneous</artifactId>
- </dependency>
- <dependency>
- <groupId>org.scala-lang</groupId>
- <artifactId>scala-library</artifactId>
- <version>${scala.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.spark</groupId>
- <artifactId>spark-core_${scala.binary.version}</artifactId>
- <version>${spark4.version}</version>
- <scope>provided</scope>
- <exclusions>
- <exclusion>
- <groupId>org.scala-lang</groupId>
- <artifactId>scala-library</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.scala-lang</groupId>
- <artifactId>scalap</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.google.code.findbugs</groupId>
- <artifactId>jsr305</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.xerial.snappy</groupId>
- <artifactId>snappy-java</artifactId>
- </exclusion>
- <exclusion>
- <groupId>xerces</groupId>
- <artifactId>xercesImpl</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-client-api</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-client-runtime</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.spark</groupId>
- <artifactId>spark-sql_${scala.binary.version}</artifactId>
- <version>${spark4.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase-client</artifactId>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase-common</artifactId>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.hbase.connectors.spark</groupId>
- <artifactId>hbase-spark-protocol-shaded</artifactId>
- <version>${revision}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.yetus</groupId>
- <artifactId>audience-annotations</artifactId>
- </dependency>
- </dependencies>
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase.thirdparty</groupId>
+ <artifactId>hbase-shaded-miscellaneous</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <version>${scala.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-core_${scala.binary.version}</artifactId>
+ <version>${spark4.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scalap</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.xerial.snappy</groupId>
+ <artifactId>snappy-java</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>xerces</groupId>
+ <artifactId>xercesImpl</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-runtime</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <version>${spark4.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase.connectors.spark</groupId>
+ <artifactId>hbase-spark-protocol-shaded</artifactId>
+ <version>${revision}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.yetus</groupId>
+ <artifactId>audience-annotations</artifactId>
+ </dependency>
+ </dependencies>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-compiler-plugin</artifactId>
- </plugin>
- <plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>build-helper-maven-plugin</artifactId>
- <executions>
- <execution>
- <id>add-shared-pushdown-sources-java</id>
- <phase>validate</phase>
- <goals>
- <goal>add-source</goal>
- </goals>
- <configuration>
- <sources>
-
<source>${project.basedir}/../../spark/hbase-spark-pushdown/src/main/java</source>
-
<source>${project.basedir}/../../spark/hbase-spark-pushdown/src/main/scala</source>
- </sources>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-enforcer-plugin</artifactId>
- <executions>
- <execution>
- <id>banned-scala</id>
- <goals>
- <goal>enforce</goal>
- </goals>
- <configuration>
- <skip>true</skip>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-checkstyle-plugin</artifactId>
- </plugin>
- <plugin>
- <groupId>net.alchim31.maven</groupId>
- <artifactId>scala-maven-plugin</artifactId>
- <version>4.9.1</version>
- <configuration>
- <charset>${project.build.sourceEncoding}</charset>
- <scalaVersion>${scala.version}</scalaVersion>
- <!-- Java depends on Scala; compile Scala before javac
sees .java sources. -->
- <compileOrder>ScalaThenJava</compileOrder>
- <!-- Use javac from maven-compiler-plugin so
protocol-shaded.jar is on the classpath (Zinc
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-shared-pushdown-sources-java</id>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <phase>validate</phase>
+ <configuration>
+ <sources>
+
<source>${project.basedir}/../../spark/hbase-spark-pushdown/src/main/java</source>
+
<source>${project.basedir}/../../spark/hbase-spark-pushdown/src/main/scala</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-enforcer-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>banned-scala</id>
+ <goals>
+ <goal>enforce</goal>
+ </goals>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ <version>4.9.1</version>
+ <configuration>
+ <charset>${project.build.sourceEncoding}</charset>
+ <scalaVersion>${scala.version}</scalaVersion>
+ <!-- Java depends on Scala; compile Scala before javac sees .java
sources. -->
+ <compileOrder>ScalaThenJava</compileOrder>
+ <!-- Use javac from maven-compiler-plugin so protocol-shaded.jar is
on the classpath (Zinc
integration missed it for translated source roots).
-->
- <sendJavaToScalac>false</sendJavaToScalac>
- <args>
- <arg>-feature</arg>
- <arg>-release</arg>
- <arg>17</arg>
- </args>
- <source>17</source>
- <target>17</target>
- </configuration>
- <executions>
- <execution>
- <id>scala-compile-first</id>
- <goals>
- <goal>compile</goal>
- </goals>
- <phase>process-resources</phase>
- </execution>
- <execution>
- <id>scala-test-compile</id>
- <goals>
- <goal>testCompile</goal>
- </goals>
- <phase>process-test-resources</phase>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>net.revelc.code</groupId>
- <artifactId>warbucks-maven-plugin</artifactId>
- <version>1.1.0</version>
- <configuration>
- <ignoreRuleFailures>true</ignoreRuleFailures>
- <rules>
- <rule>
-
<classPattern>(?!.*(.generated.|.tmpl.|\$|PushdownMappedField|SparkSQLPushDownFilter|org.apache.hadoop.hbase.spark.hbase.package)).*</classPattern>
- <includeTestClasses>false</includeTestClasses>
- <includePublicClasses>true</includePublicClasses>
-
<includePackagePrivateClasses>false</includePackagePrivateClasses>
-
<includeProtectedClasses>false</includeProtectedClasses>
-
<includePrivateClasses>false</includePrivateClasses>
-
<classAnnotationPattern>org[.]apache[.]yetus[.]audience[.]InterfaceAudience.*</classAnnotationPattern>
- </rule>
- </rules>
- </configuration>
- </plugin>
- </plugins>
- </build>
+ <sendJavaToScalac>false</sendJavaToScalac>
+ <args>
+ <arg>-feature</arg>
+ <arg>-release</arg>
+ <arg>17</arg>
+ </args>
+ <source>17</source>
+ <target>17</target>
+ </configuration>
+ <executions>
+ <execution>
+ <id>scala-compile-first</id>
+ <goals>
+ <goal>compile</goal>
+ </goals>
+ <phase>process-resources</phase>
+ </execution>
+ <execution>
+ <id>scala-test-compile</id>
+ <goals>
+ <goal>testCompile</goal>
+ </goals>
+ <phase>process-test-resources</phase>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>net.revelc.code</groupId>
+ <artifactId>warbucks-maven-plugin</artifactId>
+ <version>1.1.0</version>
+ <configuration>
+ <ignoreRuleFailures>true</ignoreRuleFailures>
+ <rules>
+ <rule>
+
<classPattern>(?!.*(.generated.|.tmpl.|\$|PushdownMappedField|SparkSQLPushDownFilter|org.apache.hadoop.hbase.spark.hbase.package)).*</classPattern>
+ <includeTestClasses>false</includeTestClasses>
+ <includePublicClasses>true</includePublicClasses>
+
<includePackagePrivateClasses>false</includePackagePrivateClasses>
+ <includeProtectedClasses>false</includeProtectedClasses>
+ <includePrivateClasses>false</includePrivateClasses>
+
<classAnnotationPattern>org[.]apache[.]yetus[.]audience[.]InterfaceAudience.*</classAnnotationPattern>
+ </rule>
+ </rules>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
</project>
diff --git a/spark4/hbase-spark4/pom.xml b/spark4/hbase-spark4/pom.xml
index 7ab30be..ee9dff9 100644
--- a/spark4/hbase-spark4/pom.xml
+++ b/spark4/hbase-spark4/pom.xml
@@ -94,6 +94,20 @@
<artifactId>spark-sql_${scala.binary.version}</artifactId>
<version>${spark4.version}</version>
<scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-runtime</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
@@ -110,6 +124,66 @@
<groupId>org.apache.yetus</groupId>
<artifactId>audience-annotations</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-mapreduce</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-shaded-testing-util</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <version>${hadoop.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-minikdc</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>javax.validation</groupId>
+ <artifactId>validation-api</artifactId>
+ <version>2.0.1.Final</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ <version>2.5.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest_${scala.binary.version}</artifactId>
+ <version>${spark4.scalatest.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
@@ -200,7 +274,30 @@
</execution>
<execution>
<id>scala-test-compile</id>
- <phase>none</phase>
+ <goals>
+ <goal>testCompile</goal>
+ </goals>
+ <phase>process-test-resources</phase>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest-maven-plugin</artifactId>
+ <version>${spark4.scalatest.maven.plugin.version}</version>
+ <configuration>
+
<reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
+ <junitxml>.</junitxml>
+ <filereports>WDF TestSuite.txt</filereports>
+ <parallel>false</parallel>
+ </configuration>
+ <executions>
+ <execution>
+ <id>test</id>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <phase>test</phase>
</execution>
</executions>
</plugin>
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala
new file mode 100644
index 0000000..da66ed6
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala
@@ -0,0 +1,170 @@
+/*
+ * 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.conf.Configuration
+import org.apache.hadoop.hbase.{CellUtil, TableName}
+import org.apache.hadoop.hbase.client.{Result, Scan}
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable
+import org.apache.hadoop.hbase.mapreduce.{IdentityTableMapper,
TableInputFormat, TableMapReduceUtil}
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hadoop.mapreduce.Job
+import org.apache.hadoop.security.UserGroupInformation
+import org.apache.spark.{SerializableWritable, SparkContext}
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.Row
+import org.apache.yetus.audience.InterfaceAudience
+import scala.reflect.ClassTag
+
+/**
+ * Narrow Spark 4 port of `org.apache.hadoop.hbase.spark.HBaseContext`: enough
for datasource scans
+ * via `NewHBaseRDD` / TableInputFormat. Bulk load, streaming, and imperative
helpers are intentionally
+ * omitted for now and will be implemented later (see HBASE-30178).
+ */
[email protected]
+class HBaseContext(@transient val sc: SparkContext, @transient val config:
Configuration)
+ extends Serializable
+ with Logging {
+
+ @transient protected var tmpHdfsConfiguration: Configuration = config
+
+ {
+ val j = Job.getInstance(config)
+ TableMapReduceUtil.initCredentials(j)
+ }
+
+ val broadcastedConf: Broadcast[SerializableWritable[Configuration]] =
+ sc.broadcast(new SerializableWritable(config))
+
+ LatestHBaseContextCache.latest = this
+
+ /**
+ * Produces an RDD of type U by scanning an HBase table and applying a
transformation function
+ * to each (key, result) pair.
+ *
+ * @param tableName the name of the HBase table to scan
+ * @param scan the Scan object defining the read parameters
+ * @param f transformation function applied to each
(ImmutableBytesWritable, Result) pair
+ * @tparam U the element type of the resulting RDD
+ * @return RDD containing the transformed scan results
+ */
+ def hbaseRDD[U: ClassTag](
+ tableName: TableName,
+ scan: Scan,
+ f: ((ImmutableBytesWritable, Result)) => U): RDD[U] = {
+ applyMapFunction(buildNewHBaseRDD(tableName, scan), f)
+ }
+
+ private def buildNewHBaseRDD(
+ tableName: TableName,
+ scan: Scan): NewHBaseRDD[ImmutableBytesWritable, Result] = {
+
+ val mapJob = Job.getInstance(getConf(broadcastedConf))
+ TableMapReduceUtil.initCredentials(mapJob)
+ TableMapReduceUtil.initTableMapperJob(
+ tableName,
+ scan,
+ classOf[IdentityTableMapper],
+ null,
+ null,
+ mapJob)
+
+ val jconf = new JobConf(mapJob.getConfiguration)
+ val jobCreds = jconf.getCredentials()
+ UserGroupInformation.setConfiguration(sc.hadoopConfiguration)
+ jobCreds.mergeAll(UserGroupInformation.getCurrentUser().getCredentials())
+
+ new NewHBaseRDD(
+ sc,
+ classOf[TableInputFormat],
+ classOf[ImmutableBytesWritable],
+ classOf[Result],
+ mapJob.getConfiguration,
+ this)
+ }
+
+ private def applyMapFunction[U: ClassTag](
+ rdd: RDD[(ImmutableBytesWritable, Result)],
+ f: ((ImmutableBytesWritable, Result)) => U): RDD[U] = {
+ rdd.map(f)
+ }
+
+ /**
+ * Produces an RDD of raw (ImmutableBytesWritable, Result) pairs by scanning
an HBase table.
+ *
+ * @param tableName the name of the HBase table to scan
+ * @param scans the Scan object defining the read parameters
+ * @return RDD of (ImmutableBytesWritable, Result) pairs
+ */
+ def hbaseRDD(tableName: TableName, scans: Scan):
RDD[(ImmutableBytesWritable, Result)] = {
+ hbaseRDD[(ImmutableBytesWritable, Result)](
+ tableName,
+ scans,
+ (r: (ImmutableBytesWritable, Result)) => r)
+ }
+
+ /**
+ * Scans an HBase table and returns an RDD of Spark SQL Rows, extracting the
specified columns
+ * from each result. This method handles cell-to-Row conversion internally,
avoiding Spark 4
+ * closure serialization issues that arise when client code maps over raw
HBase results in
+ * spark-shell.
+ *
+ * @param tableName the name of the HBase table to scan
+ * @param scan the Scan object defining the read parameters
+ * @param columns ordered sequence of column qualifiers to extract from
each row
+ * @return RDD[Row] with one field per column (null if a column is
absent in a result)
+ */
+ def hbaseRDDAsRows(tableName: String, scan: Scan, columns: Seq[String]):
RDD[Row] = {
+ val rdd = buildNewHBaseRDD(TableName.valueOf(tableName), scan)
+ val cols = columns.toArray
+ rdd.map {
+ case (_, result) =>
+ val cells = result.listCells()
+ val values = new java.util.HashMap[String, String](cells.size())
+ val iter = cells.iterator()
+ while (iter.hasNext) {
+ val cell = iter.next()
+ values.put(
+ Bytes.toString(CellUtil.cloneQualifier(cell)),
+ Bytes.toString(CellUtil.cloneValue(cell)))
+ }
+ Row.fromSeq(cols.map(col => values.get(col): Any))
+ }
+ }
+
+ private[hbase] def getConf(
+ configBroadcast: Broadcast[SerializableWritable[Configuration]]):
Configuration = {
+
+ if (tmpHdfsConfiguration == null) {
+ try {
+ tmpHdfsConfiguration = configBroadcast.value.value
+ } catch {
+ case ex: Exception => logError("Unable to getConfig from broadcast",
ex)
+ }
+ }
+ tmpHdfsConfiguration
+ }
+}
+
+/** Same semantics as Spark 3: last constructed `HBaseContext` wins. */
[email protected]
+object LatestHBaseContextCache {
+ var latest: HBaseContext = null
+}
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/Logging.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/Logging.scala
new file mode 100644
index 0000000..ec199e7
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/Logging.scala
@@ -0,0 +1,101 @@
+/*
+ * 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.yetus.audience.InterfaceAudience
+import org.slf4j.{Logger, LoggerFactory}
+
+/**
+ * Mirrors `org.apache.hadoop.hbase.spark.Logging` in `hbase-spark` for
Spark-independent SLF4J use.
+ */
[email protected]
+trait Logging {
+
+ @transient private var underlyingLogger: Logger = null
+
+ protected def logName: String =
+ this.getClass.getName.stripSuffix("$")
+
+ protected def log: Logger = {
+ if (underlyingLogger == null) {
+ initializeLogIfNecessary(false)
+ underlyingLogger = LoggerFactory.getLogger(logName)
+ }
+ underlyingLogger
+ }
+
+ protected def logInfo(msg: => String): Unit = {
+ if (log.isInfoEnabled) log.info(msg)
+ }
+
+ protected def logDebug(msg: => String): Unit = {
+ if (log.isDebugEnabled) log.debug(msg)
+ }
+
+ protected def logTrace(msg: => String): Unit = {
+ if (log.isTraceEnabled) log.trace(msg)
+ }
+
+ protected def logWarning(msg: => String): Unit = {
+ if (log.isWarnEnabled) log.warn(msg)
+ }
+
+ protected def logError(msg: => String): Unit = {
+ if (log.isErrorEnabled) log.error(msg)
+ }
+
+ protected def logInfo(msg: => String, throwable: Throwable): Unit = {
+ if (log.isInfoEnabled) log.info(msg, throwable)
+ }
+
+ protected def logDebug(msg: => String, throwable: Throwable): Unit = {
+ if (log.isDebugEnabled) log.debug(msg, throwable)
+ }
+
+ protected def logTrace(msg: => String, throwable: Throwable): Unit = {
+ if (log.isTraceEnabled) log.trace(msg, throwable)
+ }
+
+ protected def logWarning(msg: => String, throwable: Throwable): Unit = {
+ if (log.isWarnEnabled) log.warn(msg, throwable)
+ }
+
+ protected def logError(msg: => String, throwable: Throwable): Unit = {
+ if (log.isErrorEnabled) log.error(msg, throwable)
+ }
+
+ protected def initializeLogIfNecessary(isInterpreter: Boolean): Unit = {
+ if (!Logging.initialized) {
+ Logging.initLock.synchronized {
+ if (!Logging.initialized) {
+ initializeLogging(isInterpreter)
+ }
+ }
+ }
+ }
+
+ private def initializeLogging(isInterpreter: Boolean): Unit = {
+ Logging.initialized = true
+ }
+
+}
+
+private object Logging {
+ @volatile private var initialized = false
+ val initLock = new Object()
+}
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/Spark4Placeholder.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/NewHBaseRDD.scala
similarity index 54%
rename from
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/Spark4Placeholder.scala
rename to
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/NewHBaseRDD.scala
index f6f96cf..de536fd 100644
---
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/Spark4Placeholder.scala
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/NewHBaseRDD.scala
@@ -15,14 +15,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.hadoop.hbase.spark
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.mapreduce.InputFormat
+import org.apache.spark.{InterruptibleIterator, Partition, SparkContext,
TaskContext}
+import org.apache.spark.rdd.NewHadoopRDD
import org.apache.yetus.audience.InterfaceAudience
-/**
- * Phase-0 build marker for the Spark 4 connector line (`hbase-spark4`).
- * Replaced incrementally during feature porting; see umbrella jira
HBASE-30178.
- */
[email protected]
-private[hbase] object Spark4Placeholder
[email protected]
+class NewHBaseRDD[K, V](
+ @transient val sc: SparkContext,
+ @transient val inputFormatClass: Class[_ <: InputFormat[K, V]],
+ @transient val keyClass: Class[K],
+ @transient val valueClass: Class[V],
+ @transient private val __conf: Configuration,
+ val hBaseContext: HBaseContext)
+ extends NewHadoopRDD(sc, inputFormatClass, keyClass, valueClass, __conf) {
+
+ override def compute(theSplit: Partition, context: TaskContext):
InterruptibleIterator[(K, V)] = {
+ super.compute(theSplit, context)
+ }
+}
diff --git
a/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/HBaseContextSuite.scala
b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/HBaseContextSuite.scala
new file mode 100644
index 0000000..d6a4b98
--- /dev/null
+++
b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/HBaseContextSuite.scala
@@ -0,0 +1,161 @@
+/*
+ * 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.{CellUtil, HBaseTestingUtility, TableName}
+import org.apache.hadoop.hbase.client.{ConnectionFactory, Put, Scan}
+import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.spark.{SparkConf, SparkContext}
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.funsuite.AnyFunSuite
+
+class HBaseContextSuite extends AnyFunSuite with BeforeAndAfterAll with
Logging {
+
+ @transient var sc: SparkContext = _
+ var hbaseContext: HBaseContext = _
+ val TEST_UTIL = new HBaseTestingUtility
+
+ val tableName = "t1"
+ val columnFamily = "c"
+
+ override def beforeAll(): Unit = {
+ TEST_UTIL.startMiniCluster()
+ logInfo(" - minicluster started")
+
+ TEST_UTIL.createTable(TableName.valueOf(tableName),
Bytes.toBytes(columnFamily))
+ logInfo(" - created table " + tableName)
+
+ populateTestData()
+
+ val sparkConf = new SparkConf()
+ .setMaster("local[2]")
+ .setAppName("HBaseContextSuite")
+ .set("spark.hadoopRDD.ignoreEmptySplits", "false")
+ sc = new SparkContext(sparkConf)
+
+ hbaseContext = new HBaseContext(sc, TEST_UTIL.getConfiguration)
+ }
+
+ override def afterAll(): Unit = {
+ if (sc != null) {
+ sc.stop()
+ }
+ TEST_UTIL.shutdownMiniCluster()
+ TEST_UTIL.cleanupTestDir()
+ }
+
+ private def populateTestData(): Unit = {
+ val connection =
ConnectionFactory.createConnection(TEST_UTIL.getConfiguration)
+ val table = connection.getTable(TableName.valueOf(tableName))
+ try {
+ val puts = Array(
+ makePut("row1", "a", "value1"),
+ makePut("row2", "a", "value2"),
+ makePut("row2", "b", "value2b"),
+ makePut("row3", "a", "value3"),
+ makePut("row4", "a", "value4"),
+ makePut("row5", "a", "value5"))
+ puts.foreach(table.put)
+ } finally {
+ table.close()
+ connection.close()
+ }
+ }
+
+ private def makePut(rowKey: String, qualifier: String, value: String): Put =
{
+ val put = new Put(Bytes.toBytes(rowKey))
+ put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier),
Bytes.toBytes(value))
+ put
+ }
+
+ test("hbaseRDD with map function transforms scan results") {
+ val scan = new Scan()
+ val cf = Bytes.toBytes(columnFamily)
+ val qual = Bytes.toBytes("a")
+
+ val rdd = hbaseContext.hbaseRDD[String](
+ TableName.valueOf(tableName),
+ scan,
+ (r: (ImmutableBytesWritable, org.apache.hadoop.hbase.client.Result)) => {
+ val key = Bytes.toString(r._1.copyBytes())
+ val value =
Bytes.toString(CellUtil.cloneValue(r._2.getColumnLatestCell(cf, qual)))
+ s"$key=$value"
+ })
+
+ val results = rdd.collect().sorted
+ assert(results.length == 5)
+ assert(results.contains("row1=value1"))
+ assert(results.contains("row2=value2"))
+ assert(results.contains("row3=value3"))
+ assert(results.contains("row4=value4"))
+ assert(results.contains("row5=value5"))
+ }
+
+ test("hbaseRDD raw scan returns correct results with start/stop row") {
+ val scan = new Scan()
+ scan.withStartRow(Bytes.toBytes("row2"))
+ scan.withStopRow(Bytes.toBytes("row4"))
+
+ val rdd = hbaseContext.hbaseRDD(TableName.valueOf(tableName), scan)
+
+ val results = rdd.map(r =>
Bytes.toString(r._1.copyBytes())).collect().sorted
+ assert(results.length == 2)
+ assert(results(0) == "row2")
+ assert(results(1) == "row3")
+ }
+
+ test("hbaseRDD with FirstKeyOnlyFilter limits cells per row") {
+ val scan = new Scan()
+ scan.withStartRow(Bytes.toBytes("row2"))
+ scan.withStopRow(Bytes.toBytes("row3"))
+ scan.setFilter(new FirstKeyOnlyFilter())
+
+ val rdd = hbaseContext.hbaseRDD(TableName.valueOf(tableName), scan)
+
+ val cellCount = rdd.map(r => r._2.listCells().size()).collect().sum
+ // row2 has 2 cells (a, b) but FirstKeyOnlyFilter returns only 1 per row
+ assert(cellCount == 1)
+ }
+
+ test("hbaseRDDAsRows extracts specified columns into Row objects") {
+ val scan = new Scan()
+ scan.withStartRow(Bytes.toBytes("row1"))
+ scan.withStopRow(Bytes.toBytes("row2"))
+
+ val rdd = hbaseContext.hbaseRDDAsRows(tableName, scan, Seq("a"))
+
+ val rows = rdd.collect()
+ assert(rows.length == 1)
+ assert(rows(0).getString(0) == "value1")
+ }
+
+ test("hbaseRDDAsRows returns null for missing columns") {
+ val scan = new Scan()
+ scan.withStartRow(Bytes.toBytes("row1"))
+ scan.withStopRow(Bytes.toBytes("row2"))
+
+ val rdd = hbaseContext.hbaseRDDAsRows(tableName, scan, Seq("a",
"nonexistent"))
+
+ val rows = rdd.collect()
+ assert(rows.length == 1)
+ assert(rows(0).getString(0) == "value1")
+ assert(rows(0).get(1) == null)
+ }
+}
diff --git a/spark4/pom.xml b/spark4/pom.xml
index 80f6135..336049a 100644
--- a/spark4/pom.xml
+++ b/spark4/pom.xml
@@ -28,13 +28,11 @@
</parent>
<artifactId>spark4</artifactId>
- <packaging>pom</packaging>
<version>${revision}</version>
+ <packaging>pom</packaging>
<name>Apache HBase - Spark 4</name>
- <description>
- Maven reactor for the Apache Spark 4.x HBase connector (parallel to
spark/).
- Activated with profile -Pspark4 on the repository root pom. Requires JDK
17+ (Spark 4.x).
- </description>
+ <description>Maven reactor for the Apache Spark 4.x HBase connector
(parallel to spark/).
+ Activated with profile -Pspark4 on the repository root pom. Requires JDK
17+ (Spark 4.x).</description>
<modules>
<module>hbase-spark-pushdown</module>
@@ -69,10 +67,8 @@
<rules>
<requireJavaVersion>
<version>[17,)</version>
- <message>
- The Spark 4 connector reactor requires JDK 17 or newer
(Apache Spark 4.x).
- Omit -Pspark4 to build only the Kafka and Spark 3 modules
on JDK 8+.
- </message>
+ <message>The Spark 4 connector reactor requires JDK 17 or
newer (Apache Spark 4.x).
+ Omit -Pspark4 to build only the Kafka and Spark 3 modules
on JDK 8+.</message>
</requireJavaVersion>
</rules>
</configuration>
