Copilot commented on code in PR #162: URL: https://github.com/apache/hbase-connectors/pull/162#discussion_r3978688204
########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseScanBuilder.scala: ########## @@ -0,0 +1,102 @@ +/* + * 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.spark.Logging +import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownFilters, SupportsPushDownRequiredColumns} +import org.apache.spark.sql.sources._ +import org.apache.spark.sql.types.StructType +import org.apache.yetus.audience.InterfaceAudience +import scala.collection.mutable.ListBuffer + +/** + * This is a new class in the spark4 module. + * Implements ScanBuilder, SupportsPushDownFilters, and SupportsPushDownRequiredColumns. + * This is where Catalyst negotiates with the connector. Spark calls pushFilters() with + * candidate predicates and the builder accepts what it can handle and returns the rest. + * Spark calls pruneColumns() to say which columns it actually needs. + * Then build() produces the final scan plan. + * + * In the spark 3 V1 model, this negotiation happened implicitly via + * PrunedFilteredScan.buildScan(requiredColumns, filters) as a single method call with no back-and-forth. + * + * @param schema + * @param properties + */ [email protected] +class HBaseScanBuilder(schema: StructType, properties: Map[String, String]) + extends ScanBuilder + with SupportsPushDownFilters + with SupportsPushDownRequiredColumns + with Logging { + + private val catalog = HBaseTableCatalog(properties) + private val encoderClsName = + properties.getOrElse(HBaseSparkConf.QUERY_ENCODER, HBaseSparkConf.DEFAULT_QUERY_ENCODER) + @transient private val encoder = JavaBytesEncoder.create(encoderClsName) + + private var _pushedFilters: Array[Filter] = Array.empty + private var requiredSchema: StructType = schema + + override def pushFilters(filters: Array[Filter]): Array[Filter] = { + val usePushDown = properties + .get(HBaseSparkConf.PUSHDOWN_COLUMN_FILTER) + .map(_.toBoolean) + .getOrElse(HBaseSparkConf.DEFAULT_PUSHDOWN_COLUMN_FILTER) + + if (!usePushDown) { + _pushedFilters = Array.empty + return filters + } + + def isSupported(f: Filter): Boolean = f match { + case EqualTo(attr, _) => catalog.sMap.map.contains(attr) + case LessThan(attr, _) => catalog.sMap.map.contains(attr) + case GreaterThan(attr, _) => catalog.sMap.map.contains(attr) + case LessThanOrEqual(attr, _) => catalog.sMap.map.contains(attr) + case GreaterThanOrEqual(attr, _) => catalog.sMap.map.contains(attr) + case StringStartsWith(attr, _) => catalog.sMap.map.contains(attr) + case IsNull(attr) => catalog.sMap.map.contains(attr) + case IsNotNull(attr) => catalog.sMap.map.contains(attr) Review Comment: Composite row-key fields are reported as fully pushed, but the read path treats each field value as the complete HBase key. For a catalog with `rowkey: "key1:key2"`, `EqualTo("key1", "A")` plans `Get(bytes("A"))`, so every actual composite key is skipped; the server filter also maps the whole key to one component. Until component-aware encoding/filtering is implemented, predicates on composite row-key fields must be returned to Spark as unsupported. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseBatch.scala: ########## @@ -0,0 +1,120 @@ +/* + * 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.fs.Path +import org.apache.hadoop.hbase.{HBaseConfiguration, TableName} +import org.apache.hadoop.hbase.spark.{HBaseConnectionCache, Logging} +import org.apache.spark.sql.connector.read.{Batch, InputPartition, PartitionReaderFactory} +import org.apache.spark.sql.sources._ +import org.apache.spark.sql.types.StructType +import org.apache.yetus.audience.InterfaceAudience + +/** + * This is a new class in the spark4 module. Implements Batch. + * Responsible for physical planning: splits the read into partitions. + * Calls RegionLocator.getStartEndKeys() to discover HBase regions, + * intersects them with the row key filter's scan ranges, and produces an array of InputPartition objects. + * + * In the spark 3 DS V1 model, this logic was inside HBaseTableScanRDD.getPartitions(). + * + * @param requiredSchema + * @param properties + * @param catalog + * @param rowKeyFilter + * @param pushedFilters + * @param encoderClsName + */ [email protected] +class HBaseBatch( + requiredSchema: StructType, + properties: Map[String, String], + catalog: HBaseTableCatalog, + rowKeyFilter: RowKeyFilter, + pushedFilters: Array[Filter], + encoderClsName: String) + extends Batch + with Logging { + + override def planInputPartitions(): Array[InputPartition] = { + val conf = HBaseConfiguration.create() + val configResources = properties.get(HBaseSparkConf.HBASE_CONFIG_LOCATION) + configResources.foreach(_.split(",").foreach(r => conf.addResource(new Path(r)))) Review Comment: This creates HBase configuration from classpath defaults only, unlike the existing Spark 3 path, which layers `HBaseConfiguration` over `SparkContext.hadoopConfiguration`. Consequently deployments configured through `spark.hadoop.hbase.*` will ignore their quorum/authentication settings and may connect to the default cluster unless they also provide a separate config file option. Capture the Spark Hadoop configuration on the driver and pass a serializable form to both planning and readers. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReader.scala: ########## @@ -0,0 +1,424 @@ +/* + * 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.util.ArrayList +import org.apache.hadoop.fs.Path +import org.apache.hadoop.hbase.{CellUtil, HBaseConfiguration, TableName} +import org.apache.hadoop.hbase.client.{Get, Query, Result, ResultScanner, Scan, Table} +import org.apache.hadoop.hbase.spark.{AndLogicExpression, DynamicLogicExpression, + EqualLogicExpression, GreaterThanLogicExpression, GreaterThanOrEqualLogicExpression, + HBaseConnectionCache, IsNullLogicExpression, LessThanLogicExpression, + LessThanOrEqualLogicExpression, Logging, OrLogicExpression, PassThroughLogicExpression, + PushdownMappedField, SmartConnection, SparkSQLPushDownFilter, StartsWithLogicExpression} +import org.apache.hadoop.hbase.util.Bytes +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.GenericInternalRow +import org.apache.spark.sql.catalyst.util.DateTimeUtils +import org.apache.spark.sql.types.Decimal +import org.apache.spark.sql.connector.read.PartitionReader +import org.apache.spark.sql.sources._ +import org.apache.spark.sql.types._ +import org.apache.spark.unsafe.types.UTF8String +import org.apache.yetus.audience.InterfaceAudience +import scala.collection.mutable.ListBuffer +import scala.jdk.CollectionConverters._ + +/** + * This is a new class in the spark4 module. Extends PartitionReader[InternalRow] for reading data from HBase regions. + * The actual execution: opens an HBase scanner on the partition's range, attaches the SparkSQLPushDownFilter, + * reads Result objects, and converts them to InternalRow. Implements next()/get()/close(). + * + * + * In the spark 3 DS V1 model, this logic was inside DefaultSource.buildScan() + * which returned an RDD[Row] with its own compute() method. + * + * Ranges are executed as Scan operations, whilst points are executed as batched Get operations. This mirrors the spark3 + * HBaseTableScanRDD.compute() behavior. + */ [email protected] +class HBasePartitionReader( + partition: HBaseInputPartition, + requiredSchema: StructType, + properties: Map[String, String], + catalog: HBaseTableCatalog, + pushedFilters: Array[Filter], + encoderClsName: String, + usePushDownColumnFilter: Boolean) + extends PartitionReader[InternalRow] + with Logging { + + private val conf = HBaseConfiguration.create() + private val configResources = properties.get(HBaseSparkConf.HBASE_CONFIG_LOCATION) + configResources.foreach(_.split(",").foreach(r => conf.addResource(new Path(r)))) + + private val connection: SmartConnection = HBaseConnectionCache.getConnection(conf) + private val tableName = s"${catalog.namespace}:${catalog.name}" + private val table: Table = connection.getTable(TableName.valueOf(tableName)) + + private val requiredFields = requiredSchema.fieldNames.map(catalog.sMap.getField(_)) + private val filterFields = extractFilterFields(pushedFilters) + private val scanFields = (requiredFields ++ filterFields).distinct.filterNot(_.isRowKey) + private val pushDownFilter: Option[SparkSQLPushDownFilter] = buildPushDownFilter() + + private val bulkGetSize = properties + .get(HBaseSparkConf.BULKGET_SIZE) + .map(_.toInt) + .getOrElse(HBaseSparkConf.DEFAULT_BULKGET_SIZE) + + private val blockCacheEnable = properties + .get(HBaseSparkConf.QUERY_CACHEBLOCKS) + .map(_.toBoolean) + .getOrElse(HBaseSparkConf.DEFAULT_QUERY_CACHEBLOCKS) + + private val scanners = new ListBuffer[ResultScanner]() + + private val resultIterator: Iterator[Result] = { + val scanIterators = partition.scanRanges.map { range => + val scanner = buildScanner(range) + scanners += scanner + scannerToIterator(scanner) + } + val getIterator = if (partition.points.nonEmpty) { + buildGets(partition.points) + } else { + Iterator.empty + } + scanIterators.foldLeft(Iterator.empty: Iterator[Result])(_ ++ _) ++ getIterator + } + + private var currentResult: Result = _ + + override def next(): Boolean = { + if (resultIterator.hasNext) { + currentResult = resultIterator.next() + true + } else { + false + } + } + + private val keyFields = catalog.getRowKey + + override def get(): InternalRow = { + val rowKey = currentResult.getRow + + val keyValues = parseRowKey(rowKey, keyFields) + val values = new Array[Any](requiredFields.length) + + requiredFields.zipWithIndex.foreach { case (field, idx) => + if (field.isRowKey) { + values(idx) = convertToInternalRow(keyValues.get(field).orNull, field.dt) + } else { + val cell = currentResult.getColumnLatestCell( + Bytes.toBytes(field.cf), Bytes.toBytes(field.col)) + if (cell == null || cell.getValueLength == 0) { + values(idx) = null Review Comment: A present zero-length HBase value is not null: it is a valid empty string or empty binary value. This branch converts both to SQL null, so projections and null predicates become inconsistent. Only treat a missing cell as null and pass zero-length values through the normal decoder. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseInputPartition.scala: ########## @@ -0,0 +1,35 @@ +/* + * 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.connector.read.InputPartition +import org.apache.yetus.audience.InterfaceAudience + +/** + * This is a new class in the spark4 module. Implements InputPartition for serialization of the partition + * information to be sent to executors. + * + * Ranges are executed as HBase Scan operations; points as batched Get operations. + * This mirrors the spark3 HBaseScanPartition behavior. + */ [email protected] +case class HBaseInputPartition( + index: Int, + scanRanges: Seq[Range], + points: Seq[Array[Byte]]) + extends InputPartition Review Comment: These partitions expose no preferred locations, so Spark schedules every HBase scan without region-server locality. The Spark 3 path explicitly returns each partition's region server from `getPreferredLocations`; dropping that information can turn distributed scans into cross-node RPC traffic. Carry the region host from planning and override `preferredLocations()` here. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/ScanRange.scala: ########## @@ -0,0 +1,222 @@ +/* + * 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.yetus.audience.InterfaceAudience +import scala.collection.mutable.ListBuffer + +/** + * This is a new class in the spark4 module. Wraps ScanRange and RowKeyFilter. + * + * Extracted from DefaultSource.scala in spark3 where it was an inner class. Handles merging scan ranges + * from row key predicates (union/intersect). Same logic, just in its own file now for clarity. + * + * @param upperBound + * @param isUpperBoundEqualTo + * @param lowerBound + * @param isLowerBoundEqualTo + */ + [email protected] +class ScanRange( + var upperBound: Array[Byte], + var isUpperBoundEqualTo: Boolean, + var lowerBound: Array[Byte], + var isLowerBoundEqualTo: Boolean) + extends Serializable { + + def mergeIntersect(other: ScanRange): Unit = { + val upperBoundCompare = compareRange(upperBound, other.upperBound) + val lowerBoundCompare = compareRange(lowerBound, other.lowerBound) + + upperBound = if (upperBoundCompare < 0) upperBound else other.upperBound + lowerBound = if (lowerBoundCompare > 0) lowerBound else other.lowerBound + + isLowerBoundEqualTo = + if (lowerBoundCompare == 0) + isLowerBoundEqualTo && other.isLowerBoundEqualTo + else if (lowerBoundCompare < 0) other.isLowerBoundEqualTo + else isLowerBoundEqualTo + + isUpperBoundEqualTo = + if (upperBoundCompare == 0) + isUpperBoundEqualTo && other.isUpperBoundEqualTo + else if (upperBoundCompare < 0) isUpperBoundEqualTo + else other.isUpperBoundEqualTo + } + + def mergeUnion(other: ScanRange): Unit = { + val upperBoundCompare = compareRange(upperBound, other.upperBound) + val lowerBoundCompare = compareRange(lowerBound, other.lowerBound) + + upperBound = if (upperBoundCompare > 0) upperBound else other.upperBound + lowerBound = if (lowerBoundCompare < 0) lowerBound else other.lowerBound + + isLowerBoundEqualTo = + if (lowerBoundCompare == 0) + isLowerBoundEqualTo || other.isLowerBoundEqualTo + else if (lowerBoundCompare < 0) isLowerBoundEqualTo + else other.isLowerBoundEqualTo + + isUpperBoundEqualTo = + if (upperBoundCompare == 0) + isUpperBoundEqualTo || other.isUpperBoundEqualTo + else if (upperBoundCompare < 0) other.isUpperBoundEqualTo + else isUpperBoundEqualTo + } + + def getOverLapScanRange(other: ScanRange): ScanRange = { + var leftRange: ScanRange = null + var rightRange: ScanRange = null + + if (compareRange(lowerBound, other.lowerBound) < 0 || + compareRange(upperBound, other.upperBound) < 0) { + leftRange = this + rightRange = other + } else { + leftRange = other + rightRange = this + } + + if (hasOverlap(leftRange, rightRange)) { + val result = new ScanRange(upperBound, isUpperBoundEqualTo, lowerBound, isLowerBoundEqualTo) + result.mergeIntersect(other) + result + } else { + null + } + } + + def hasOverlap(left: ScanRange, right: ScanRange): Boolean = { + val cmp = compareRange(left.upperBound, right.lowerBound) + if (cmp > 0) true + else if (cmp == 0) left.isUpperBoundEqualTo && right.isLowerBoundEqualTo + else false + } + + def compareRange(left: Array[Byte], right: Array[Byte]): Int = { + if (left == null && right == null) 0 + else if (left == null && right != null) 1 + else if (left != null && right == null) -1 + else Bytes.compareTo(left, right) + } + + def containsPoint(point: Array[Byte]): Boolean = { + val lowerCompare = compareRange(point, lowerBound) + val upperCompare = compareRange(point, upperBound) + + ((isLowerBoundEqualTo && lowerCompare >= 0) || + (!isLowerBoundEqualTo && lowerCompare > 0)) && + ((isUpperBoundEqualTo && upperCompare <= 0) || + (!isUpperBoundEqualTo && upperCompare < 0)) + } + + override def toString: String = { + "ScanRange:(upperBound:" + Bytes.toString(upperBound) + + ",isUpperBoundEqualTo:" + isUpperBoundEqualTo + ",lowerBound:" + + Bytes.toString(lowerBound) + ",isLowerBoundEqualTo:" + isLowerBoundEqualTo + ")" + } +} + [email protected] +class RowKeyFilter( + currentPoint: Array[Byte] = null, + currentRange: ScanRange = new ScanRange(null, true, new Array[Byte](0), true), + var points: ListBuffer[Array[Byte]] = new ListBuffer[Array[Byte]](), + var ranges: ListBuffer[ScanRange] = new ListBuffer[ScanRange]()) + extends Serializable { + + if (currentRange != null) ranges += currentRange + if (currentPoint != null) points += currentPoint + + def mergeUnion(other: RowKeyFilter): RowKeyFilter = { + other.points.foreach(p => points += p) + + other.ranges.foreach { otherR => + var doesOverLap = false + ranges.foreach { r => + if (r.getOverLapScanRange(otherR) != null) { + r.mergeUnion(otherR) + doesOverLap = true + } + } + if (!doesOverLap) ranges += otherR Review Comment: Unioning points by appending them without reconciling them with existing ranges produces duplicate output. For example, `key < 'm' OR key = 'a'` leaves both the `< m` scan and an `a` Get, and the reader concatenates both results, returning row `a` twice. Deduplicate points and remove any point covered by a unioned range; range merging should likewise leave disjoint normalized ranges. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseBatch.scala: ########## @@ -0,0 +1,120 @@ +/* + * 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.fs.Path +import org.apache.hadoop.hbase.{HBaseConfiguration, TableName} +import org.apache.hadoop.hbase.spark.{HBaseConnectionCache, Logging} +import org.apache.spark.sql.connector.read.{Batch, InputPartition, PartitionReaderFactory} +import org.apache.spark.sql.sources._ +import org.apache.spark.sql.types.StructType +import org.apache.yetus.audience.InterfaceAudience + +/** + * This is a new class in the spark4 module. Implements Batch. + * Responsible for physical planning: splits the read into partitions. + * Calls RegionLocator.getStartEndKeys() to discover HBase regions, + * intersects them with the row key filter's scan ranges, and produces an array of InputPartition objects. + * + * In the spark 3 DS V1 model, this logic was inside HBaseTableScanRDD.getPartitions(). + * + * @param requiredSchema + * @param properties + * @param catalog + * @param rowKeyFilter + * @param pushedFilters + * @param encoderClsName + */ [email protected] +class HBaseBatch( + requiredSchema: StructType, + properties: Map[String, String], + catalog: HBaseTableCatalog, + rowKeyFilter: RowKeyFilter, + pushedFilters: Array[Filter], + encoderClsName: String) + extends Batch + with Logging { + + override def planInputPartitions(): Array[InputPartition] = { + val conf = HBaseConfiguration.create() + val configResources = properties.get(HBaseSparkConf.HBASE_CONFIG_LOCATION) + configResources.foreach(_.split(",").foreach(r => conf.addResource(new Path(r)))) + + val connection = HBaseConnectionCache.getConnection(conf) + try { + val tableName = s"${catalog.namespace}:${catalog.name}" + val regionLocator = connection.getRegionLocator(TableName.valueOf(tableName)) + try { + val keys = regionLocator.getStartEndKeys + val startKeys = keys.getFirst + val endKeys = keys.getSecond + + val regions = startKeys.zip(endKeys).zipWithIndex.map { case ((start, end), idx) => + HBaseRegion(idx, Some(start), Some(end)) + } + + val scanRanges = rowKeyFilter.ranges.toSeq + val points = rowKeyFilter.points.toSeq + + if (scanRanges.isEmpty && points.isEmpty) { + regions.map { region => + val fullRange = Range(region) + HBaseInputPartition(region.index, Seq(fullRange), Seq.empty): InputPartition + } Review Comment: An empty row-key filter represents an unsatisfiable intersection, not an unfiltered query: the no-filter case already contains the default unbounded range. Predicates such as `key < 'a' AND key > 'z'` therefore trigger a full-table scan here before the server filter discards every row. Return no input partitions when both collections are empty. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReader.scala: ########## @@ -0,0 +1,424 @@ +/* + * 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.util.ArrayList +import org.apache.hadoop.fs.Path +import org.apache.hadoop.hbase.{CellUtil, HBaseConfiguration, TableName} +import org.apache.hadoop.hbase.client.{Get, Query, Result, ResultScanner, Scan, Table} +import org.apache.hadoop.hbase.spark.{AndLogicExpression, DynamicLogicExpression, + EqualLogicExpression, GreaterThanLogicExpression, GreaterThanOrEqualLogicExpression, + HBaseConnectionCache, IsNullLogicExpression, LessThanLogicExpression, + LessThanOrEqualLogicExpression, Logging, OrLogicExpression, PassThroughLogicExpression, + PushdownMappedField, SmartConnection, SparkSQLPushDownFilter, StartsWithLogicExpression} +import org.apache.hadoop.hbase.util.Bytes +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.GenericInternalRow +import org.apache.spark.sql.catalyst.util.DateTimeUtils +import org.apache.spark.sql.types.Decimal +import org.apache.spark.sql.connector.read.PartitionReader +import org.apache.spark.sql.sources._ +import org.apache.spark.sql.types._ +import org.apache.spark.unsafe.types.UTF8String +import org.apache.yetus.audience.InterfaceAudience +import scala.collection.mutable.ListBuffer +import scala.jdk.CollectionConverters._ + +/** + * This is a new class in the spark4 module. Extends PartitionReader[InternalRow] for reading data from HBase regions. + * The actual execution: opens an HBase scanner on the partition's range, attaches the SparkSQLPushDownFilter, + * reads Result objects, and converts them to InternalRow. Implements next()/get()/close(). + * + * + * In the spark 3 DS V1 model, this logic was inside DefaultSource.buildScan() + * which returned an RDD[Row] with its own compute() method. + * + * Ranges are executed as Scan operations, whilst points are executed as batched Get operations. This mirrors the spark3 + * HBaseTableScanRDD.compute() behavior. + */ [email protected] +class HBasePartitionReader( + partition: HBaseInputPartition, + requiredSchema: StructType, + properties: Map[String, String], + catalog: HBaseTableCatalog, + pushedFilters: Array[Filter], + encoderClsName: String, + usePushDownColumnFilter: Boolean) + extends PartitionReader[InternalRow] + with Logging { + + private val conf = HBaseConfiguration.create() + private val configResources = properties.get(HBaseSparkConf.HBASE_CONFIG_LOCATION) + configResources.foreach(_.split(",").foreach(r => conf.addResource(new Path(r)))) + + private val connection: SmartConnection = HBaseConnectionCache.getConnection(conf) + private val tableName = s"${catalog.namespace}:${catalog.name}" + private val table: Table = connection.getTable(TableName.valueOf(tableName)) + + private val requiredFields = requiredSchema.fieldNames.map(catalog.sMap.getField(_)) + private val filterFields = extractFilterFields(pushedFilters) + private val scanFields = (requiredFields ++ filterFields).distinct.filterNot(_.isRowKey) + private val pushDownFilter: Option[SparkSQLPushDownFilter] = buildPushDownFilter() + + private val bulkGetSize = properties + .get(HBaseSparkConf.BULKGET_SIZE) + .map(_.toInt) + .getOrElse(HBaseSparkConf.DEFAULT_BULKGET_SIZE) + + private val blockCacheEnable = properties + .get(HBaseSparkConf.QUERY_CACHEBLOCKS) + .map(_.toBoolean) + .getOrElse(HBaseSparkConf.DEFAULT_QUERY_CACHEBLOCKS) + + private val scanners = new ListBuffer[ResultScanner]() + + private val resultIterator: Iterator[Result] = { + val scanIterators = partition.scanRanges.map { range => + val scanner = buildScanner(range) + scanners += scanner + scannerToIterator(scanner) + } + val getIterator = if (partition.points.nonEmpty) { + buildGets(partition.points) + } else { + Iterator.empty + } + scanIterators.foldLeft(Iterator.empty: Iterator[Result])(_ ++ _) ++ getIterator + } + + private var currentResult: Result = _ + + override def next(): Boolean = { + if (resultIterator.hasNext) { + currentResult = resultIterator.next() + true + } else { + false + } + } + + private val keyFields = catalog.getRowKey + + override def get(): InternalRow = { + val rowKey = currentResult.getRow + + val keyValues = parseRowKey(rowKey, keyFields) + val values = new Array[Any](requiredFields.length) + + requiredFields.zipWithIndex.foreach { case (field, idx) => + if (field.isRowKey) { + values(idx) = convertToInternalRow(keyValues.get(field).orNull, field.dt) + } else { + val cell = currentResult.getColumnLatestCell( + Bytes.toBytes(field.cf), Bytes.toBytes(field.col)) + if (cell == null || cell.getValueLength == 0) { + values(idx) = null + } else { + val v = CellUtil.cloneValue(cell) + val scalaValue = field.dt match { + case BinaryType => v + case _ => Utils.hbaseFieldToScalaType(field, v, 0, v.length) + } + values(idx) = convertToInternalRow(scalaValue, field.dt) + } + } + } + new GenericInternalRow(values) + } + + override def close(): Unit = { + scanners.foreach(s => if (s != null) s.close()) + if (table != null) table.close() + if (connection != null) connection.close() + } + + private def setStopRow(scan: Scan, bound: Bound): Scan = { + if (bound.inc) { + val incremented = Utils.incrementByteArray(bound.b) + if (incremented != null) scan.withStopRow(incremented) + else scan + } else { + scan.withStopRow(bound.b) + } + } + + private def buildScanner(range: Range): ResultScanner = { + val scan = (range.lower, range.upper) match { + case (Some(Bound(a, _)), Some(upper)) => + setStopRow(new Scan().withStartRow(a), upper) + case (None, Some(upper)) => + setStopRow(new Scan(), upper) + case (Some(Bound(a, _)), None) => + new Scan().withStartRow(a) + case (None, None) => + new Scan() + } + + scan.setCacheBlocks(blockCacheEnable) + properties.get(HBaseSparkConf.QUERY_CACHEDROWS).map(_.toInt).foreach { rows => + if (rows > 0) scan.setCaching(rows) + } + properties.get(HBaseSparkConf.QUERY_BATCHSIZE).map(_.toInt).foreach { batch => + if (batch > 0) scan.setBatch(batch) + } + handleTimeSemantics(scan) + + scanFields.foreach { f => + scan.addColumn(f.cfBytes, f.colBytes) + } Review Comment: Restricting the scan to the qualifier tested by `IsNull` drops exactly the rows that should match when that qualifier is absent. For example, `select(key).filter(name IS NULL)` scans only `cf:name`; an HBase row with other cells but no `name` produces no `Result`, so the filter never evaluates it. When a pushed expression can match a missing qualifier, scan enough data to preserve row existence (the safe fallback is no column restriction). This issue also appears on line 196 of the same file. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReader.scala: ########## @@ -0,0 +1,424 @@ +/* + * 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.util.ArrayList +import org.apache.hadoop.fs.Path +import org.apache.hadoop.hbase.{CellUtil, HBaseConfiguration, TableName} +import org.apache.hadoop.hbase.client.{Get, Query, Result, ResultScanner, Scan, Table} +import org.apache.hadoop.hbase.spark.{AndLogicExpression, DynamicLogicExpression, + EqualLogicExpression, GreaterThanLogicExpression, GreaterThanOrEqualLogicExpression, + HBaseConnectionCache, IsNullLogicExpression, LessThanLogicExpression, + LessThanOrEqualLogicExpression, Logging, OrLogicExpression, PassThroughLogicExpression, + PushdownMappedField, SmartConnection, SparkSQLPushDownFilter, StartsWithLogicExpression} +import org.apache.hadoop.hbase.util.Bytes +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.GenericInternalRow +import org.apache.spark.sql.catalyst.util.DateTimeUtils +import org.apache.spark.sql.types.Decimal +import org.apache.spark.sql.connector.read.PartitionReader +import org.apache.spark.sql.sources._ +import org.apache.spark.sql.types._ +import org.apache.spark.unsafe.types.UTF8String +import org.apache.yetus.audience.InterfaceAudience +import scala.collection.mutable.ListBuffer +import scala.jdk.CollectionConverters._ + +/** + * This is a new class in the spark4 module. Extends PartitionReader[InternalRow] for reading data from HBase regions. + * The actual execution: opens an HBase scanner on the partition's range, attaches the SparkSQLPushDownFilter, + * reads Result objects, and converts them to InternalRow. Implements next()/get()/close(). + * + * + * In the spark 3 DS V1 model, this logic was inside DefaultSource.buildScan() + * which returned an RDD[Row] with its own compute() method. + * + * Ranges are executed as Scan operations, whilst points are executed as batched Get operations. This mirrors the spark3 + * HBaseTableScanRDD.compute() behavior. + */ [email protected] +class HBasePartitionReader( + partition: HBaseInputPartition, + requiredSchema: StructType, + properties: Map[String, String], + catalog: HBaseTableCatalog, + pushedFilters: Array[Filter], + encoderClsName: String, + usePushDownColumnFilter: Boolean) + extends PartitionReader[InternalRow] + with Logging { + + private val conf = HBaseConfiguration.create() + private val configResources = properties.get(HBaseSparkConf.HBASE_CONFIG_LOCATION) + configResources.foreach(_.split(",").foreach(r => conf.addResource(new Path(r)))) + + private val connection: SmartConnection = HBaseConnectionCache.getConnection(conf) + private val tableName = s"${catalog.namespace}:${catalog.name}" + private val table: Table = connection.getTable(TableName.valueOf(tableName)) + + private val requiredFields = requiredSchema.fieldNames.map(catalog.sMap.getField(_)) + private val filterFields = extractFilterFields(pushedFilters) + private val scanFields = (requiredFields ++ filterFields).distinct.filterNot(_.isRowKey) + private val pushDownFilter: Option[SparkSQLPushDownFilter] = buildPushDownFilter() + + private val bulkGetSize = properties + .get(HBaseSparkConf.BULKGET_SIZE) + .map(_.toInt) + .getOrElse(HBaseSparkConf.DEFAULT_BULKGET_SIZE) + + private val blockCacheEnable = properties + .get(HBaseSparkConf.QUERY_CACHEBLOCKS) + .map(_.toBoolean) + .getOrElse(HBaseSparkConf.DEFAULT_QUERY_CACHEBLOCKS) + + private val scanners = new ListBuffer[ResultScanner]() + + private val resultIterator: Iterator[Result] = { + val scanIterators = partition.scanRanges.map { range => + val scanner = buildScanner(range) + scanners += scanner + scannerToIterator(scanner) + } + val getIterator = if (partition.points.nonEmpty) { + buildGets(partition.points) + } else { + Iterator.empty + } + scanIterators.foldLeft(Iterator.empty: Iterator[Result])(_ ++ _) ++ getIterator + } + + private var currentResult: Result = _ + + override def next(): Boolean = { + if (resultIterator.hasNext) { + currentResult = resultIterator.next() + true + } else { + false + } + } + + private val keyFields = catalog.getRowKey + + override def get(): InternalRow = { + val rowKey = currentResult.getRow + + val keyValues = parseRowKey(rowKey, keyFields) + val values = new Array[Any](requiredFields.length) + + requiredFields.zipWithIndex.foreach { case (field, idx) => + if (field.isRowKey) { + values(idx) = convertToInternalRow(keyValues.get(field).orNull, field.dt) + } else { + val cell = currentResult.getColumnLatestCell( + Bytes.toBytes(field.cf), Bytes.toBytes(field.col)) + if (cell == null || cell.getValueLength == 0) { + values(idx) = null + } else { + val v = CellUtil.cloneValue(cell) + val scalaValue = field.dt match { + case BinaryType => v + case _ => Utils.hbaseFieldToScalaType(field, v, 0, v.length) + } + values(idx) = convertToInternalRow(scalaValue, field.dt) + } + } + } + new GenericInternalRow(values) + } + + override def close(): Unit = { + scanners.foreach(s => if (s != null) s.close()) + if (table != null) table.close() + if (connection != null) connection.close() + } + + private def setStopRow(scan: Scan, bound: Bound): Scan = { + if (bound.inc) { + val incremented = Utils.incrementByteArray(bound.b) + if (incremented != null) scan.withStopRow(incremented) + else scan Review Comment: `incrementByteArray` is a numeric successor, not the immediate lexicographic successor of an arbitrary variable-length row key. For an inclusive upper bound `"a"`, this sets stop row to `"b"` and scans every `"a..."` key even though only the exact `"a"` key is in range; an all-`0xff` bound becomes unbounded. Append a zero byte, as the Spark 3 path does, to form the exclusive stop row immediately after the exact key. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
