You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by co...@apache.org on 2024/11/29 17:27:12 UTC
(hudi) branch master updated: [HUDI-8604] Support index command with External table in Spark SQL (#12372)
This is an automated email from the ASF dual-hosted git repository.
codope 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 6a726bf840c [HUDI-8604] Support index command with External table in Spark SQL (#12372)
6a726bf840c is described below
commit 6a726bf840cd7934e8a863f3a381d1187e1f111b
Author: Lokesh Jain <lj...@apache.org>
AuthorDate: Fri Nov 29 22:57:01 2024 +0530
[HUDI-8604] Support index command with External table in Spark SQL (#12372)
* [HUDI-8604] Support index command with External table in Spark SQL
* Fix compilation failure
---
.../spark/sql/hudi/command/index/TestExpressionIndex.scala | 12 +++++++++---
.../spark/sql/hudi/command/index/TestSecondaryIndex.scala | 10 +++++++++-
.../scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala | 7 ++++---
.../scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala | 6 ++++--
.../scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala | 8 +++++---
5 files changed, 31 insertions(+), 12 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/command/index/TestExpressionIndex.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/command/index/TestExpressionIndex.scala
index 9d3f7b24414..f09ae398e0e 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/command/index/TestExpressionIndex.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/command/index/TestExpressionIndex.scala
@@ -68,7 +68,7 @@ class TestExpressionIndex extends HoodieSparkSqlTestBase {
initQueryIndexConf()
}
- test("Test Expression Index With Hive Sync Non Partitioned Table") {
+ test("Test Expression Index With Hive Sync Non Partitioned External Table") {
// There is a big difference between Java class loader architecture of versions 1.8 and 17.
// Hive 2.3.7 is compiled with Java 1.8, and the class loader used there throws error when Hive APIs are run on Java 17.
// So we special case this test only for Java 8.
@@ -96,12 +96,16 @@ class TestExpressionIndex extends HoodieSparkSqlTestBase {
""".stripMargin)
// ts=1000 and from_unixtime(ts, 'yyyy-MM-dd') = '1970-01-01'
spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+
+ spark.sql(s"""DROP TABLE if exists $tableName""")
+ // Use the same base path as above
+ spark.sql(s"""CREATE TABLE $tableName USING hudi LOCATION '$basePath'""")
+ // create expression index
+ spark.sql(s"""create index idx_datestr on $tableName using column_stats(ts) options(expr='from_unixtime', format='yyyy-MM-dd HH:mm');""")
// ts=100000 and from_unixtime(ts, 'yyyy-MM-dd') = '1970-01-02'
spark.sql(s"insert into $tableName values(2, 'a2', 10, 100000)")
// ts=10000000 and from_unixtime(ts, 'yyyy-MM-dd') = '1970-04-26'
spark.sql(s"insert into $tableName values(3, 'a3', 10, 10000000)")
- // create expression index
- spark.sql(s"create index idx_datestr on $tableName using column_stats(ts) options(expr='from_unixtime', format='yyyy-MM-dd HH:mm')")
val metaClient = createMetaClient(spark, basePath)
assertTrue(metaClient.getIndexMetadata.isPresent)
val expressionIndexMetadata = metaClient.getIndexMetadata.get()
@@ -139,6 +143,8 @@ class TestExpressionIndex extends HoodieSparkSqlTestBase {
hiveClient.close()
tool.close()
HiveTestUtil.shutdown()
+
+ spark.sql(s"""DROP TABLE if exists $tableName""")
}
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/command/index/TestSecondaryIndex.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/command/index/TestSecondaryIndex.scala
index b2d81bed774..342cdbf5b6d 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/command/index/TestSecondaryIndex.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/command/index/TestSecondaryIndex.scala
@@ -59,7 +59,7 @@ class TestSecondaryIndex extends HoodieSparkSqlTestBase {
DataSourceWriteOptions.RECORD_MERGE_MODE.key() -> RecordMergeMode.COMMIT_TIME_ORDERING.name()
) ++ metadataOpts
- test("Test Create/Show/Drop Secondary Index") {
+ test("Test Create/Show/Drop Secondary Index with External Table") {
if (HoodieSparkUtils.gteqSpark3_3) {
withTempDir { tmp =>
Seq("cow", "mor").foreach { tableType =>
@@ -86,6 +86,14 @@ class TestSecondaryIndex extends HoodieSparkSqlTestBase {
| location '$basePath'
""".stripMargin)
spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+
+ spark.sql(s"""DROP TABLE if exists $tableName""")
+ // Use the same base path as above
+ spark.sql(
+ s"""CREATE TABLE $tableName USING hudi options (
+ | hoodie.metadata.record.index.enable = 'true'
+ | ) LOCATION '$basePath'""".stripMargin)
+
spark.sql(s"insert into $tableName values(2, 'a2', 10, 1001)")
spark.sql(s"insert into $tableName values(3, 'a3', 10, 1002)")
checkAnswer(s"show indexes from default.$tableName")(
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
index a053646c13e..35afd221abe 100644
--- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
@@ -18,20 +18,19 @@
package org.apache.spark.sql.adapter
import org.apache.hudi.Spark33HoodieFileScanRDD
-
import org.apache.avro.Schema
import org.apache.hadoop.conf.Configuration
import org.apache.spark.sql._
import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
+import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression}
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.util.METADATA_COL_ATTR_KEY
-import org.apache.spark.sql.connector.catalog.V2TableWithV1Fallback
+import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, Spark33LegacyHoodieParquetFileFormat, Spark33ParquetReader, SparkParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
@@ -57,6 +56,8 @@ class Spark3_3Adapter extends BaseSpark3Adapter {
// on top of it
case PhysicalOperation(_, _, DataSourceV2Relation(v2: V2TableWithV1Fallback, _, _, _, _)) if isHoodieTable(v2.v1Table) =>
Some(v2.v1Table)
+ case ResolvedTable(_, _, V1Table(v1Table), _) if isHoodieTable(v1Table) =>
+ Some(v1Table)
case _ => None
}
}
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
index 1e2807df55b..1eb6c97f362 100644
--- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
@@ -22,14 +22,14 @@ import org.apache.hadoop.conf.Configuration
import org.apache.hudi.Spark34HoodieFileScanRDD
import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
+import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression}
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.util.METADATA_COL_ATTR_KEY
-import org.apache.spark.sql.connector.catalog.V2TableWithV1Fallback
+import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, Spark34LegacyHoodieParquetFileFormat, Spark34ParquetReader, SparkParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.execution.datasources._
@@ -56,6 +56,8 @@ class Spark3_4Adapter extends BaseSpark3Adapter {
// on top of it
case PhysicalOperation(_, _, DataSourceV2Relation(v2: V2TableWithV1Fallback, _, _, _, _)) if isHoodieTable(v2.v1Table) =>
Some(v2.v1Table)
+ case ResolvedTable(_, _, V1Table(v1Table), _) if isHoodieTable(v1Table) =>
+ Some(v1Table)
case _ => None
}
}
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
index 6b07bc7690a..40019f3f28a 100644
--- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
@@ -23,16 +23,16 @@ import org.apache.hudi.Spark35HoodieFileScanRDD
import org.apache.spark.sql._
import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
+import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression}
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.util.METADATA_COL_ATTR_KEY
-import org.apache.spark.sql.connector.catalog.V2TableWithV1Fallback
+import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
import org.apache.spark.sql.execution.datasources._
-import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, Spark35ParquetReader, Spark35LegacyHoodieParquetFileFormat, SparkParquetReader}
+import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, Spark35LegacyHoodieParquetFileFormat, Spark35ParquetReader, SparkParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
import org.apache.spark.sql.internal.SQLConf
@@ -56,6 +56,8 @@ class Spark3_5Adapter extends BaseSpark3Adapter {
// on top of it
case PhysicalOperation(_, _, DataSourceV2Relation(v2: V2TableWithV1Fallback, _, _, _, _)) if isHoodieTable(v2) =>
Some(v2.v1Table)
+ case ResolvedTable(_, _, V1Table(v1Table), _) if isHoodieTable(v1Table) =>
+ Some(v1Table)
case _ => None
}
}