Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[AMORO-3387]fix: create table like using mixed_hive #3387

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
+38 −27
Open

[AMORO-3387]fix: create table like using mixed_hive #3387

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
49538d5
fix: create table like using mixed_hive
Aireed Jan 2, 2025
e346ab5
fix: create table like using mixed_hive
Aireed Jan 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 38 additions & 27 deletions ...c/main/scala/org/apache/amoro/spark/sql/catalyst/analysis/RewriteMixedFormatCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.connector.catalog.TableCatalog
import org.apache.spark.sql.execution.command.CreateTableLikeCommand

import org.apache.amoro.spark.{MixedFormatSparkCatalog, MixedFormatSparkSessionCatalog}
import org.apache.amoro.spark.{MixedFormatSparkCatalog, MixedFormatSparkSessionCatalog, SparkUnifiedCatalog, SparkUnifiedSessionCatalog}
import org.apache.amoro.spark.mixed.MixedSessionCatalogBase
import org.apache.amoro.spark.sql.MixedFormatExtensionUtils.buildCatalogAndIdentifier
import org.apache.amoro.spark.sql.catalyst.plans.{AlterMixedFormatTableDropPartition, TruncateMixedFormatTable}
Expand All @@ -51,6 +51,10 @@ case class RewriteMixedFormatCommand(sparkSession: SparkSession) extends Rule[Lo
case _: MixedFormatSparkSessionCatalog[_] =>
provider.isDefined && MixedSessionCatalogBase.SUPPORTED_PROVIDERS.contains(
provider.get.toLowerCase)
case _: SparkUnifiedCatalog => true
case _: SparkUnifiedSessionCatalog[_] =>
provider.isDefined && MixedSessionCatalogBase.SUPPORTED_PROVIDERS.contains(
provider.get.toLowerCase)
case _ => false
}
}
Expand Down Expand Up @@ -82,35 +86,42 @@ case class RewriteMixedFormatCommand(sparkSession: SparkSession) extends Rule[Lo
optionsMap += (WriteMode.WRITE_MODE_KEY -> WriteMode.OVERWRITE_DYNAMIC.mode)
val newTableSpec = tableSpec.copy(properties = propertiesMap)
c.copy(tableSpec = newTableSpec, writeOptions = optionsMap)
case CreateTableLikeCommand(targetTable, sourceTable, _, provider, properties, ifNotExists)
if isCreateMixedFormatTableLikeCommand(targetTable, provider) =>
val (sourceCatalog, sourceIdentifier) = buildCatalogAndIdentifier(sparkSession, sourceTable)
val (targetCatalog, targetIdentifier) = buildCatalogAndIdentifier(sparkSession, targetTable)
val table = sourceCatalog.loadTable(sourceIdentifier)
var targetProperties = properties
targetProperties += ("provider" -> "arctic")
table match {
case keyedTable: MixedSparkTable =>
keyedTable.table() match {
case table: KeyedTable =>
targetProperties += ("primary.keys" -> String.join(
",",
table.primaryKeySpec().fieldNames()))
case CreateTableLikeCommand(targetTable, sourceTable, _, provider, properties, ifNotExists) =>
isCreateMixedFormatTableLikeCommand(targetTable, provider) match {
case true =>
val (sourceCatalog, sourceIdentifier) =
buildCatalogAndIdentifier(sparkSession, sourceTable)
val (targetCatalog, targetIdentifier) =
buildCatalogAndIdentifier(sparkSession, targetTable)
val table = sourceCatalog.loadTable(sourceIdentifier)
var targetProperties = properties
targetProperties += ("provider" -> "mixed_hive")
table match {
case keyedTable: MixedSparkTable =>
keyedTable.table() match {
case table: KeyedTable =>
targetProperties += ("primary.keys" -> String.join(
",",
table.primaryKeySpec().fieldNames()))
case _ =>
}
case _ =>
}
case _ =>
val tableSpec = TableSpec(
properties = targetProperties.toMap,
provider = provider,
options = Map.empty,
location = None,
comment = None,
serde = None,
external = false)
val seq: Seq[String] = Seq(
targetTable.database.getOrElse(sparkSession.catalog.currentDatabase),
targetTable.identifier)
val name = ResolvedDBObjectName(targetCatalog, seq)
CreateTable(name, table.schema(), table.partitioning(), tableSpec, ifNotExists)
case _ => plan
}
val tableSpec = TableSpec(
properties = targetProperties.toMap,
provider = provider,
options = Map.empty,
location = None,
comment = None,
serde = None,
external = false)
val seq: Seq[String] = Seq(targetTable.database.get, targetTable.identifier)
val name = ResolvedDBObjectName(targetCatalog, seq)
CreateTable(name, table.schema(), table.partitioning(), tableSpec, ifNotExists)
case _ => plan
}
}
Expand Down
Loading