Skip to content

[Spark] Fix the dynamic partion issue #5897

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

Jump to bottom
Open
openinx wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[Spark] Fix the dynamic partion issue #5897

openinx wants to merge 2 commits into from

Conversation

@openinx
Copy link
Collaborator

@openinx openinx commented Jan 21, 2026

Which Delta project/connector is this regarding?

  • Spark
  • Standalone
  • Flink
  • Kernel
  • Other (fill in here)

Description

My step to reproduce this bug

Step.1 Create the table and insert data into the UC managed table

Since we cannot create table via the external OSS-Stack currently.

CREATE TABLE main.demo_zh.test_catalog_owned_commits (
 time TIMESTAMP NOT NULL,
 time_date_level DATE,
 time_hour_level INT,
 tenant STRING,
 eventMetaId INT)
USING delta
PARTITIONED BY (time_date_level, time_hour_level)
TBLPROPERTIES (
 'delta.feature.catalogOwned-preview' = 'supported',
 'delta.autoOptimize.optimizeWrite' = 'true',
 'delta.checkpoint.writeStatsAsJson' = 'false',
 'delta.checkpoint.writeStatsAsStruct' = 'true',
 'delta.checkpointPolicy' = 'v2',
 'delta.dataSkippingNumIndexedCols' = '1',
 'delta.deletedFileRetentionDuration' = 'interval 1 days',
 'delta.enableDeletionVectors' = 'true',
 'delta.feature.deletionVectors' = 'supported',
 'delta.feature.v2Checkpoint' = 'supported',
 'delta.logRetentionDuration' = 'interval 7 days'
);
INSERT INTO main.demo_zh.test_catalog_owned_commits (
 time, time_date_level, time_hour_level, tenant, eventMetaId
)
VALUES
 ('2025-10-15T10:00:00Z', '2025-10-15', 10, 'tenant_1', 1),
 ('2025-10-15T11:00:00Z', '2025-10-15', 11, 'tenant_2', 2),
 ('2025-10-15T12:00:00Z', '2025-10-15', 12, 'tenant_3', 3);

Step.2 Use the following command to check the table is querable.

bin/spark-sql --name "local-uc-test" \
    --master "local[*]" \
    --packages "io.delta:delta-spark_2.13:4.0.0,io.unitycatalog:unitycatalog-spark_2.13:0.3.0,org.apache.hadoop:hadoop-aws:3.4.1" \
    --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
    --exclude-packages "org.apache.hadoop:hadoop-client-runtime,org.apache.hadoop:hadoop-client-api" \
    --conf "spark.sql.catalog.spark_catalog=io.unitycatalog.spark.UCSingleCatalog" \
    --conf "spark.hadoop.fs.s3.impl=org.apache.hadoop.fs.s3a.S3AFileSystem" \
    --conf "spark.sql.catalog.main=io.unitycatalog.spark.UCSingleCatalog" \
    --conf "spark.sql.catalog.main.uri=$UC_URI" \
    --conf "spark.sql.catalog.main.token=$UC_TOKEN" \
    --conf "spark.sql.defaultCatalog=main"

Execute the SQL query

select * from main.demo_zh.test_catalog_owned_commits

All seems fine.

Step.3 start the spark-shell to execute the scripts

bin/spark-shell --name "local-uc-test" \
    --master "local[*]" \
    --packages "io.delta:delta-spark_2.13:4.0.0,io.unitycatalog:unitycatalog-spark_2.13:0.3.0,org.apache.hadoop:hadoop-aws:3.4.1" \
    --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
    --exclude-packages "org.apache.hadoop:hadoop-client-runtime,org.apache.hadoop:hadoop-client-api" \
    --conf "spark.sql.catalog.spark_catalog=io.unitycatalog.spark.UCSingleCatalog" \
    --conf "spark.hadoop.fs.s3.impl=org.apache.hadoop.fs.s3a.S3AFileSystem" \
    --conf "spark.sql.catalog.main=io.unitycatalog.spark.UCSingleCatalog" \
    --conf "spark.sql.catalog.main.uri=$UC_URI" \
    --conf "spark.sql.catalog.main.token=$UC_TOKEN" \
    --conf "spark.sql.defaultCatalog=main"

And execute the following scripts

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
spark.sql("select * from main.demo_zh.test_catalog_owned_commits where time_hour_level=12")
    .limit(2)
    .write // Insert into the target table with dynamic partition overwrite
    .format("delta")
    .mode("overwrite")
    .option("partitionOverwriteMode", "dynamic")
    .saveAsTable("main.demo_zh.test_catalog_owned_commits")

Then it will report the following error:

scala> import org.apache.spark.sql.Row
     | import org.apache.spark.sql.types._
     | 
     | 
     | spark.sql("select * from main.demo_zh.test_catalog_owned_commits where time_hour_level=12")
     |     .limit(2)
     |     .write // Insert into the target table with dynamic partition overwrite
     |     .format("delta")
     |     .mode("overwrite")
     |     .option("partitionOverwriteMode", "dynamic")
     |     .saveAsTable("main.demo_zh.test_catalog_owned_commits")
25/11/11 23:08:35 WARN MetricsConfig: Cannot locate configuration: tried hadoop-metrics2-s3a-file-system.properties,hadoop-metrics2.properties
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
25/11/11 23:08:38 WARN UCCommitCoordinatorBuilder: MessageWithContext(Skipping catalog spark_catalog as it does ,{})not have both uri and token configured in Spark Session.
25/11/11 23:08:40 WARN UCCommitCoordinatorBuilder: MessageWithContext(Skipping catalog spark_catalog as it does ,{})not have both uri and token configured in Spark Session.
25/11/11 23:08:41 WARN ObjectStore: Version information not found in metastore. hive.metastore.schema.verification is not enabled so recording the schema version 2.3.0
25/11/11 23:08:41 WARN ObjectStore: setMetaStoreSchemaVersion called but recording version is disabled: version = 2.3.0, comment = Set by MetaStore [email protected]
25/11/11 23:08:41 WARN ObjectStore: Failed to get database demo_zh, returning NoSuchObjectException
org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException: [SCHEMA_NOT_FOUND] The schema `spark_catalog`.`demo_zh` cannot be found. Verify the spelling and correctness of the schema and catalog.
If you did not qualify the name with a catalog, verify the current_schema() output, or qualify the name with the correct catalog.
To tolerate the error on drop use DROP SCHEMA IF EXISTS. SQLSTATE: 42704
  at org.apache.spark.sql.catalyst.catalog.SessionCatalog.requireDbExists(SessionCatalog.scala:265)
  at org.apache.spark.sql.catalyst.catalog.SessionCatalog.getDatabaseMetadata(SessionCatalog.scala:329)
  at org.apache.spark.sql.catalyst.catalog.SessionCatalog.defaultTablePath(SessionCatalog.scala:637)
  at org.apache.spark.sql.delta.catalog.DeltaCatalog.$anonfun$createDeltaTable$10(DeltaCatalog.scala:168)
  at scala.Option.getOrElse(Option.scala:201)
  at org.apache.spark.sql.delta.catalog.DeltaCatalog.$anonfun$createDeltaTable$1(DeltaCatalog.scala:168)
  at org.apache.spark.sql.delta.metering.DeltaLogging.recordFrameProfile(DeltaLogging.scala:171)
  at org.apache.spark.sql.delta.metering.DeltaLogging.recordFrameProfile$(DeltaLogging.scala:169)
  at org.apache.spark.sql.delta.catalog.DeltaCatalog.recordFrameProfile(DeltaCatalog.scala:69)
  at org.apache.spark.sql.delta.catalog.DeltaCatalog.org$apache$spark$sql$delta$catalog$DeltaCatalog$$createDeltaTable(DeltaCatalog.scala:105)
  at org.apache.spark.sql.delta.catalog.DeltaCatalog.$anonfun$createTable$1(DeltaCatalog.scala:372)
  at org.apache.spark.sql.delta.metering.DeltaLogging.recordFrameProfile(DeltaLogging.scala:171)
  at org.apache.spark.sql.delta.metering.DeltaLogging.recordFrameProfile$(DeltaLogging.scala:169)
  at org.apache.spark.sql.delta.catalog.DeltaCatalog.recordFrameProfile(DeltaCatalog.scala:69)
  at org.apache.spark.sql.delta.catalog.DeltaCatalog.createTable(DeltaCatalog.scala:352)
  at org.apache.spark.sql.delta.catalog.DeltaCatalog.createTable(DeltaCatalog.scala:343)
  at io.unitycatalog.spark.UCSingleCatalog.createTable(UCSingleCatalog.scala:108)
  at org.apache.spark.sql.execution.datasources.v2.ReplaceTableAsSelectExec.run(WriteToDataSourceV2Exec.scala:172)
  at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result$lzycompute(V2CommandExec.scala:43)
  at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result(V2CommandExec.scala:43)
  at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.executeCollect(V2CommandExec.scala:49)
  at org.apache.spark.sql.execution.QueryExecution.$anonfun$eagerlyExecuteCommands$2(QueryExecution.scala:155)
  at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$8(SQLExecution.scala:163)
  at org.apache.spark.sql.execution.SQLExecution$.withSessionTagsApplied(SQLExecution.scala:272)
  at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$7(SQLExecution.scala:125)
  at org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)
  at org.apache.spark.sql.artifact.ArtifactManager.$anonfun$withResources$1(ArtifactManager.scala:112)
  at org.apache.spark.sql.artifact.ArtifactManager.withClassLoaderIfNeeded(ArtifactManager.scala:106)
  at org.apache.spark.sql.artifact.ArtifactManager.withResources(ArtifactManager.scala:111)
  at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$6(SQLExecution.scala:125)
  at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:295)
  at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$1(SQLExecution.scala:124)
  at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:804)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId0(SQLExecution.scala:78)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:237)
  at org.apache.spark.sql.execution.QueryExecution.$anonfun$eagerlyExecuteCommands$1(QueryExecution.scala:155)
  at org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:654)
  at org.apache.spark.sql.execution.QueryExecution.org$apache$spark$sql$execution$QueryExecution$$eagerlyExecute$1(QueryExecution.scala:154)
  at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$3.applyOrElse(QueryExecution.scala:169)
  at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$3.applyOrElse(QueryExecution.scala:164)
  at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:470)
  at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
  at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:470)
  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:37)
  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:360)
  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:356)
  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
  at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:446)
  at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:164)
  at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyCommandExecuted$1(QueryExecution.scala:126)
  at scala.util.Try$.apply(Try.scala:217)
  at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1378)
  at org.apache.spark.util.Utils$.getTryWithCallerStacktrace(Utils.scala:1439)
  at org.apache.spark.util.LazyTry.get(LazyTry.scala:58)
  at org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:131)
  at org.apache.spark.sql.execution.QueryExecution.assertCommandExecuted(QueryExecution.scala:192)
  at org.apache.spark.sql.classic.DataFrameWriter.runCommand(DataFrameWriter.scala:622)
  at org.apache.spark.sql.classic.DataFrameWriter.saveAsTable(DataFrameWriter.scala:514)
  at org.apache.spark.sql.classic.DataFrameWriter.saveAsTable(DataFrameWriter.scala:442)
  ... 42 elided
  Suppressed: org.apache.spark.util.Utils$OriginalTryStackTraceException: Full stacktrace of original doTryWithCallerStacktrace caller
    at org.apache.spark.sql.catalyst.catalog.SessionCatalog.requireDbExists(SessionCatalog.scala:265)
    at org.apache.spark.sql.catalyst.catalog.SessionCatalog.getDatabaseMetadata(SessionCatalog.scala:329)
    at org.apache.spark.sql.catalyst.catalog.SessionCatalog.defaultTablePath(SessionCatalog.scala:637)
    at org.apache.spark.sql.delta.catalog.DeltaCatalog.$anonfun$createDeltaTable$10(DeltaCatalog.scala:168)
    at scala.Option.getOrElse(Option.scala:201)
    at org.apache.spark.sql.delta.catalog.DeltaCatalog.$anonfun$createDeltaTable$1(DeltaCatalog.scala:168)
    at org.apache.spark.sql.delta.metering.DeltaLogging.recordFrameProfile(DeltaLogging.scala:171)
    at org.apache.spark.sql.delta.metering.DeltaLogging.recordFrameProfile$(DeltaLogging.scala:169)
    at org.apache.spark.sql.delta.catalog.DeltaCatalog.recordFrameProfile(DeltaCatalog.scala:69)
    at org.apache.spark.sql.delta.catalog.DeltaCatalog.org$apache$spark$sql$delta$catalog$DeltaCatalog$$createDeltaTable(DeltaCatalog.scala:105)
    at org.apache.spark.sql.delta.catalog.DeltaCatalog.$anonfun$createTable$1(DeltaCatalog.scala:372)
    at org.apache.spark.sql.delta.metering.DeltaLogging.recordFrameProfile(DeltaLogging.scala:171)
    at org.apache.spark.sql.delta.metering.DeltaLogging.recordFrameProfile$(DeltaLogging.scala:169)
    at org.apache.spark.sql.delta.catalog.DeltaCatalog.recordFrameProfile(DeltaCatalog.scala:69)
    at org.apache.spark.sql.delta.catalog.DeltaCatalog.createTable(DeltaCatalog.scala:352)
    at org.apache.spark.sql.delta.catalog.DeltaCatalog.createTable(DeltaCatalog.scala:343)
    at io.unitycatalog.spark.UCSingleCatalog.createTable(UCSingleCatalog.scala:108)
    at org.apache.spark.sql.execution.datasources.v2.ReplaceTableAsSelectExec.run(WriteToDataSourceV2Exec.scala:172)
    at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result$lzycompute(V2CommandExec.scala:43)
    at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result(V2CommandExec.scala:43)
    at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.executeCollect(V2CommandExec.scala:49)
    at org.apache.spark.sql.execution.QueryExecution.$anonfun$eagerlyExecuteCommands$2(QueryExecution.scala:155)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$8(SQLExecution.scala:163)
    at org.apache.spark.sql.execution.SQLExecution$.withSessionTagsApplied(SQLExecution.scala:272)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$7(SQLExecution.scala:125)
    at org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)
    at org.apache.spark.sql.artifact.ArtifactManager.$anonfun$withResources$1(ArtifactManager.scala:112)
    at org.apache.spark.sql.artifact.ArtifactManager.withClassLoaderIfNeeded(ArtifactManager.scala:106)
    at org.apache.spark.sql.artifact.ArtifactManager.withResources(ArtifactManager.scala:111)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$6(SQLExecution.scala:125)
    at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:295)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId0$1(SQLExecution.scala:124)
    at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:804)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId0(SQLExecution.scala:78)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:237)
    at org.apache.spark.sql.execution.QueryExecution.$anonfun$eagerlyExecuteCommands$1(QueryExecution.scala:155)
    at org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:654)
    at org.apache.spark.sql.execution.QueryExecution.org$apache$spark$sql$execution$QueryExecution$$eagerlyExecute$1(QueryExecution.scala:154)
    at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$3.applyOrElse(QueryExecution.scala:169)
    at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$3.applyOrElse(QueryExecution.scala:164)
    at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:470)
    at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:86)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:470)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:37)
    at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:360)
    at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:356)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:37)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:446)
    at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:164)
    at org.apache.spark.sql.execution.QueryExecution.$anonfun$lazyCommandExecuted$1(QueryExecution.scala:126)
    at scala.util.Try$.apply(Try.scala:217)
    at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala:1378)
    at org.apache.spark.util.LazyTry.tryT$lzycompute(LazyTry.scala:46)
    at org.apache.spark.util.LazyTry.tryT(LazyTry.scala:46)
    ... 48 more

And finally I found my table is disappeared.

How was this patch tested?

unit test.

Does this PR introduce any user-facing changes?

No

@openinx
[Spark] Fix the dynamic partion issue
2c45660 
Signed-off-by: openinx <[email protected]>
@openinx
fix
8e733f7 
Signed-off-by: openinx <[email protected]>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

1 participant

@openinx