Copilot commented on code in PR #162: URL: https://github.com/apache/hbase-connectors/pull/162#discussion_r3990808559
########## 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 + } Review Comment: The left/right range selection in `getOverLapScanRange` can be incorrect because it uses `lowerBound` OR `upperBound` comparisons. This can pick the higher-lower-bound range as `leftRange` (e.g., when one range is contained inside another), causing `hasOverlap(left, right)` to compute against the wrong ordering and potentially return `null` even when ranges overlap. Fix by ordering strictly by lower bound (and optionally tie-breaking by upper bound) so `leftRange.lower <= rightRange.lower` is always true. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseBatch.scala: ########## @@ -0,0 +1,123 @@ +/* + * 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.SparkSession +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 hadoopConf = SparkSession.active.sparkContext.hadoopConfiguration + val conf = HBaseConfiguration.create(hadoopConf) + 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), Review Comment: HBase region boundary keys from `RegionLocator.getStartEndKeys` use empty byte arrays to represent unbounded start (first region) and unbounded end (last region). Wrapping those as `Some(start)`/`Some(end)` can break range comparisons/intersections (empty array sorts before all non-empty keys) and can cause the last region (end key empty) to be treated as having an upper bound of `[]`, potentially dropping partitions or truncating scans. Convert empty start/end keys to `None` (unbounded) when building `HBaseRegion`/`Range(region)` (similar to how scan-range lower bounds are filtered with `.filter(_.nonEmpty)` below). ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReader.scala: ########## @@ -0,0 +1,435 @@ +/* + * 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.hbase.{CellUtil, 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, + wrappedConf: SerializableConfiguration) + extends PartitionReader[InternalRow] + with Logging { + + private val conf = wrappedConf.value + + 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 hasNullCheck = pushedFilters.exists(containsNullCheck) + 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 + } Review Comment: Building `resultIterator` via `foldLeft(...)(_ ++ _)` creates a chain of concatenated iterators, which can add overhead and deep nesting when `scanRanges` is large. Prefer iterating/flattening directly (e.g., flattening an iterator of iterators) to reduce iterator concatenation overhead while preserving laziness. -- 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]
