This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 308f78e71dea fix(spark): skip the catalog lookup in HoodieFileIndex
unless the read declares a bucket index (#20240)
308f78e71dea is described below
commit 308f78e71deabf7a9eec4b0a2ea26b1360658030
Author: Y Ethan Guo <[email protected]>
AuthorDate: Thu Oct 8 21:52:11 2026 -0700
fix(spark): skip the catalog lookup in HoodieFileIndex unless the read
declares a bucket index (#20240)
---
.../scala/org/apache/hudi/HoodieFileIndex.scala | 34 ++++++---
.../sql/hudi/common/TestHoodieFileIndex.scala | 85 +++++++++++++++++++++-
2 files changed, 109 insertions(+), 10 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
index 003c571acf16..eafff4e8ddb8 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
@@ -24,10 +24,12 @@ import org.apache.hudi.common.config.{HoodieConfig,
HoodieMetadataConfig, TypedP
import
org.apache.hudi.common.config.TimestampKeyGeneratorConfig.{TIMESTAMP_INPUT_DATE_FORMAT,
TIMESTAMP_OUTPUT_DATE_FORMAT}
import org.apache.hudi.common.model.{FileSlice, HoodieBaseFile, HoodieLogFile}
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
-import org.apache.hudi.common.util.StringUtils
+import org.apache.hudi.common.util.{ConfigUtils, StringUtils}
+import org.apache.hudi.config.HoodieIndexConfig
import org.apache.hudi.core.read.BaseHoodieTableFileIndex
import org.apache.hudi.core.read.BaseHoodieTableFileIndex.PartitionPath
import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.index.HoodieIndex.IndexType
import org.apache.hudi.index.bucket.partition.PartitionBucketIndexUtils
import org.apache.hudi.keygen.{TimestampBasedAvroKeyGenerator,
TimestampBasedKeyGenerator}
import org.apache.hudi.storage.{StoragePath, StoragePathInfo}
@@ -655,14 +657,21 @@ object HoodieFileIndex extends Logging {
properties.setProperty(PARTITIONPATH_FIELD.key,
HoodieTableConfig.getPartitionFieldPropForKeyGenerator(tableConfig).orElse(""))
properties.setProperty(HoodieTableConfig.PARTITION_EXTRACTOR_CLASS.key(),
tableConfig.getPartitionExtractorClass.orElse(""))
- // for simple bucket index, we need to set the INDEX_TYPE,
BUCKET_INDEX_HASH_FIELD, BUCKET_INDEX_NUM_BUCKETS
- val database = getDatabaseName(tableConfig,
spark.catalog.currentDatabase)
- val tableName = tableConfig.getTableName
-
- if (spark.catalog.tableExists(database, tableName)) {
- val tableIdentifier = TableIdentifier(tableName, Some(database))
- val table = HoodieCatalogTable(spark, tableIdentifier)
- table.catalogProperties.foreach(kv => properties.setProperty(kv._1,
kv._2))
+ // The catalog entry only completes the simple bucket index options,
which hoodie.properties does not record,
+ // so it is consulted only when the read declares a bucket index and
only those options are copied.
+ if (isBucketIndexDeclared(properties)) {
+ val database = getDatabaseName(tableConfig,
spark.catalog.currentDatabase)
+ val tableName = tableConfig.getTableName
+ try {
+ if (spark.catalog.tableExists(database, tableName)) {
+ val table = HoodieCatalogTable(spark, TableIdentifier(tableName,
Some(database)))
+ table.catalogProperties.filter { case (key, _) =>
bucketIndexCatalogKeys.contains(key) }
+ .foreach(kv => properties.setProperty(kv._1, kv._2))
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(s"Skipping catalog properties for table $tableName:
lookup of $database.$tableName failed", e)
+ }
}
}
@@ -721,6 +730,13 @@ object HoodieFileIndex extends Logging {
Seq(new StoragePath(path))
}
+ private val bucketIndexCatalogKeys: Set[String] = Set(
+ HoodieIndexConfig.INDEX_TYPE, HoodieIndexConfig.BUCKET_INDEX_ENGINE_TYPE,
HoodieIndexConfig.BUCKET_INDEX_HASH_FIELD,
+ HoodieIndexConfig.BUCKET_INDEX_NUM_BUCKETS,
HoodieIndexConfig.BUCKET_QUERY_INDEX).map(_.key)
+
+ private def isBucketIndexDeclared(properties: TypedProperties): Boolean =
+
IndexType.BUCKET.name.equalsIgnoreCase(ConfigUtils.getStringWithAltKeys(properties,
HoodieIndexConfig.INDEX_TYPE, ""))
+
// if database name is not set, fall back to use 'default' instead of failing
def getDatabaseName(tableConfig: HoodieTableConfig, defaultDatabase: String)
= {
if (StringUtils.isNullOrEmpty(tableConfig.getDatabaseName)) {
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieFileIndex.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieFileIndex.scala
index 37e23508da79..697b73b287dc 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieFileIndex.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieFileIndex.scala
@@ -21,9 +21,24 @@ package org.apache.spark.sql.hudi.common
import org.apache.hudi.HoodieFileIndex
import org.apache.hudi.common.table.HoodieTableConfig
+import org.apache.hudi.config.HoodieIndexConfig.{BUCKET_INDEX_NUM_BUCKETS,
INDEX_TYPE}
+import org.apache.hudi.index.HoodieIndex.IndexType
-import org.junit.jupiter.api.Assertions.assertEquals
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalog.Catalog
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType, SessionCatalog}
+import org.apache.spark.sql.internal.{SessionState, SQLConf}
+import org.apache.spark.sql.types.StructType
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse}
import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.io.TempDir
+import org.mockito.ArgumentMatchers.anyString
+import org.mockito.Mockito.{mock, never, verify, when}
+
+import java.net.URI
+import java.nio.file.Path
class TestHoodieFileIndex {
@Test
@@ -31,4 +46,72 @@ class TestHoodieFileIndex {
assertEquals("default", HoodieFileIndex.getDatabaseName(new
HoodieTableConfig(), null))
assertEquals("default_db", HoodieFileIndex.getDatabaseName(new
HoodieTableConfig(), "default_db"))
}
+
+ @Test
+ def testCatalogIsNotConsultedWithoutBucketIndex(): Unit = {
+ val (spark, catalog, _) = mockSpark()
+ when(catalog.tableExists(anyString(), anyString()))
+ .thenThrow(new IllegalStateException("database default is not allowed"))
+
+ val props = HoodieFileIndex.getConfigProperties(spark, Map.empty,
tableConfig("tbl"))
+
+ verify(catalog, never()).tableExists(anyString(), anyString())
+ assertFalse(props.containsKey(BUCKET_INDEX_NUM_BUCKETS.key))
+ }
+
+ @Test
+ def testCatalogPropertiesMergedWhenBucketIndexDeclared(@TempDir dir: Path):
Unit = {
+ val (spark, catalog, sessionCatalog) = mockSpark()
+ val id = TableIdentifier("tbl", Some("default"))
+ when(catalog.tableExists("default", "tbl")).thenReturn(true)
+
when(sessionCatalog.getTableMetadata(id)).thenReturn(bucketCatalogTable(id,
dir))
+ val options = Map(INDEX_TYPE.key -> IndexType.BUCKET.name)
+
+ val props = HoodieFileIndex.getConfigProperties(spark, options,
tableConfig("tbl"))
+
+ assertEquals("3", props.getProperty(BUCKET_INDEX_NUM_BUCKETS.key))
+ assertFalse(props.containsKey("preCombineField"))
+ verify(catalog).tableExists("default", "tbl")
+ }
+
+ @Test
+ def testCatalogFailureWithBucketIndexDeclaredDoesNotFailRead(): Unit = {
+ val (spark, catalog, _) = mockSpark()
+ when(catalog.tableExists(anyString(), anyString()))
+ .thenThrow(new IllegalStateException("database default is not allowed"))
+ val options = Map(INDEX_TYPE.key -> IndexType.BUCKET.name)
+
+ val props = HoodieFileIndex.getConfigProperties(spark, options,
tableConfig("tbl"))
+
+ verify(catalog).tableExists("default", "tbl")
+ assertFalse(props.containsKey(BUCKET_INDEX_NUM_BUCKETS.key))
+ }
+
+ private def tableConfig(name: String): HoodieTableConfig = {
+ val config = new HoodieTableConfig()
+ config.setValue(HoodieTableConfig.NAME, name)
+ config
+ }
+
+ private def bucketCatalogTable(id: TableIdentifier, location: Path):
CatalogTable = CatalogTable(
+ identifier = id,
+ tableType = CatalogTableType.EXTERNAL,
+ storage = CatalogStorageFormat.empty.copy(locationUri = Some(new
URI("file:" + location.toAbsolutePath))),
+ schema = new StructType(),
+ provider = Some("hudi"),
+ properties = Map(BUCKET_INDEX_NUM_BUCKETS.key -> "3", "preCombineField" ->
"ts"))
+
+ private def mockSpark(): (SparkSession, Catalog, SessionCatalog) = {
+ val spark = mock(classOf[SparkSession])
+ val sessionState = mock(classOf[SessionState])
+ val catalog = mock(classOf[Catalog])
+ val sessionCatalog = mock(classOf[SessionCatalog])
+ when(spark.sessionState).thenReturn(sessionState)
+ when(sessionState.conf).thenReturn(new SQLConf)
+ when(sessionState.newHadoopConf()).thenReturn(new Configuration())
+ when(sessionState.catalog).thenReturn(sessionCatalog)
+ when(spark.catalog).thenReturn(catalog)
+ when(catalog.currentDatabase).thenReturn("default")
+ (spark, catalog, sessionCatalog)
+ }
}
