dbt-labs / dbt-labs/dbt-adapters

[Bug] dbt seed fails for Delta Lake + S3DynamoDBLogStore configuration

Open
#476 0 comments 1 reaction 0 assignees View on GitHub
pkg:dbt-spark type:bug
Dominant language
Python
Stars
233
Forks
362
Avg merge
3d 22h
Merged PRs (30d)
9

Description

### Is this a new bug in dbt-spark?

- [X] I believe this is a new bug in dbt-spark
- [X] I have searched the existing issues, and I could not find an existing issue for this bug

### Current Behavior

`dbt seed` command produces `java.nio.file.FileSystemException: Old entries for table s3:/// still exist in the external log store` error

### Expected Behavior

`dbt seed` successfully creates table from csv seed file

### Steps To Reproduce

dbt 1.8.5 + dbt-spark 1.8.0

```yaml
seeds:
+file_format: 'delta'
```

Apache Spark configured with Delta Lake session and s3 dynamodb lock store:
```
spark.delta.logStore.s3.impl=io.delta.storage.S3DynamoDBLogStore
spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog
spark.sql.extensions=o.delta.sql.DeltaSparkSessionExtension
spark.io.delta.storage.S3DynamoDBLogStore.ddb.tableName=
```

### Relevant log output

```shell
[2024-08-20T15:50:46.627+0000] {logging_mixin.py:188} INFO - 15:50:46 Running with dbt=1.8.5
[2024-08-20T15:50:46.804+0000] {logging_mixin.py:188} INFO - 15:50:46 Registered adapter: spark=1.8.0
[2024-08-20T15:50:46.845+0000] {logging_mixin.py:188} INFO - 15:50:46 Unable to do partial parsing because saved manifest not found. Starting full parse.
[2024-08-20T15:50:49.738+0000] {logging_mixin.py:188} INFO - 15:50:49 [WARNING]: Deprecated functionality
The `tests` config has been renamed to `data_tests`. Please see
https://docs.getdbt.com/docs/build/data-tests#new-data_tests-syntax for more
information.
[2024-08-20T15:50:50.267+0000] {logging_mixin.py:188} INFO - 15:50:50 Found 6 models, 42 data tests, 1 seed, 36 sources, 733 macros, 25 unit tests
[2024-08-20T15:50:50.275+0000] {logging_mixin.py:188} INFO - 15:50:50
[2024-08-20T15:52:03.105+0000] {logging_mixin.py:188} INFO - 15:52:03 Concurrency: 4 threads (target='dev')
[2024-08-20T15:52:03.106+0000] {logging_mixin.py:188} INFO - 15:52:03
[2024-08-20T15:52:03.110+0000] {logging_mixin.py:188} INFO - 15:52:03 1 of 1 START seed file . ........... [RUN]
[2024-08-20T15:52:11.569+0000] {logging_mixin.py:188} INFO - 15:52:11 1 of 1 ERROR loading seed file . ... [ERROR in 8.45s]
[2024-08-20T15:52:11.947+0000] {logging_mixin.py:188} INFO - 15:52:11
[2024-08-20T15:52:11.948+0000] {logging_mixin.py:188} INFO - 15:52:11 Finished running 1 seed in 0 hours 1 minutes and 21.67 seconds (81.67s).
[2024-08-20T15:52:12.023+0000] {logging_mixin.py:188} INFO - 15:52:12
[2024-08-20T15:52:12.024+0000] {logging_mixin.py:188} INFO - 15:52:12 Completed with 1 error and 0 warnings:
[2024-08-20T15:52:12.025+0000] {logging_mixin.py:188} INFO - 15:52:12
[2024-08-20T15:52:12.027+0000] {logging_mixin.py:188} INFO - 15:52:12 Runtime Error in seed (seeds/.csv)
Database Error
org.apache.hive.service.cli.HiveSQLException: Error running query: java.nio.file.FileSystemException: Old entries for table still exist in the external log store
at org.apache.spark.sql.hive.thriftserver.HiveThriftServerErrors$.runningQueryError(HiveThriftServerErrors.scala:46)
at org.apache.spark.sql.hive.thriftserver.SparkExecuteStatementOperation.org$apache$spark$sql$hive$thriftserver$SparkExecuteStatementOperation$$execute(SparkExecuteStatementOperation.scala:262)
at org.apache.spark.sql.hive.thriftserver.SparkExecuteStatementOperation$$anon$2$$anon$3.$anonfun$run$2(SparkExecuteStatementOperation.scala:166)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at org.apache.spark.sql.hive.thriftserver.SparkOperation.withLocalProperties(SparkOperation.scala:79)
at org.apache.spark.sql.hive.thriftserver.SparkOperation.withLocalProperties$(SparkOperation.scala:63)
at org.apache.spark.sql.hive.thriftserver.SparkExecuteStatementOperation.withLocalProperties(SparkExecuteStatementOperation.scala:41)
at org.apache.spark.sql.hive.thriftserver.SparkExecuteStatementOperation$$anon$2$$anon$3.run(SparkExecuteStatementOperation.scala:166)
at org.apache.spark.sql.hive.thriftserver.SparkExecuteStatementOperation$$anon$2$$anon$3.run(SparkExecuteStatementOperation.scala:161)
at java.base/java.security.AccessController.doPrivileged(AccessController.java:712)
at java.base/javax.security.auth.Subject.doAs(Subject.java:439)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1899)
at org.apache.spark.sql.hive.thriftserver.SparkExecuteStatementOperation$$anon$2.run(SparkExecuteStatementOperation.scala:175)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: java.nio.file.FileSystemException: Old entries for table still exist in the external log store
at io.delta.storage.BaseExternalLogStore.write(BaseExternalLogStore.java:222)
at org.apache.spark.sql.delta.storage.LogStoreAdaptor.write(LogStore.scala:444)
at org.apache.spark.sql.delta.storage.DelegatingLogStore.write(DelegatingLogStore.scala:119)
at org.apache.spark.sql.delta.OptimisticTransactionImpl.writeCommitFile(OptimisticTransaction.scala:1806)
at org.apache.spark.sql.delta.OptimisticTransactionImpl.writeCommitFile$(OptimisticTransaction.scala:1798)
at org.apache.spark.sql.delta.OptimisticTransaction.writeCommitFile(OptimisticTransaction.scala:142)
at org.apache.spark.sql.delta.OptimisticTransactionImpl.doCommit(OptimisticTransaction.scala:1711)
at org.apache.spark.sql.delta.OptimisticTransactionImpl.doCommit$(OptimisticTransaction.scala:1682)
at org.apache.spark.sql.delta.OptimisticTransaction.doCommit(OptimisticTransaction.scala:142)
at org.apache.spark.sql.delta.OptimisticTransactionImpl.$anonfun$doCommitRetryIteratively$3(OptimisticTransaction.scala:1651)
at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:158)
at org.apache.spark.sql.delta.OptimisticTransactionImpl.$anonfun$doCommitRetryIteratively$2(OptimisticTransaction.scala:1648)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at org.apache.spark.sql.delta.metering.DeltaLogging.recordFrameProfile(DeltaLogging.scala:140)
at org.apache.spark.sql.delta.metering.DeltaLogging.recordFrameProfile$(DeltaLogging.scala:138)
at org.apache.spark.sql.delta.OptimisticTransaction.recordFrameProfile(OptimisticTransaction.scala:142)
at org.apache.spark.sql.delta.metering.DeltaLogging.$anonfun$recordDeltaOperationInternal$1(DeltaLogging.scala:133)
at com.databricks.spark.util.DatabricksLogging.recordOperation(DatabricksLogging.scala:128)
at com.databricks.spark.util.DatabricksLogging.recordOperation$(DatabricksLogging.scala:117)
at org.apache.spark.sql.delta.OptimisticTransaction.recordOperation(OptimisticTransaction.scala:142)
at org.apache.spark.sql.delta.metering.DeltaLogging.recordDeltaOperationInternal(DeltaLogging.scala:132)
```

### Environment

```markdown
- OS: Amazon Linux 2 (AWS EMR)
- Python: 3.11
- dbt-core: 1.8.5
- dbt-spark:1.8.0
```

### Additional Context

Possible root cause.
dbt-spark uses drop table + create table for seeds.
![image](https://github.com/user-attachments/assets/63beb9ee-8b33-4f6f-9e2c-077d83ad0397)

On the first seed run, everything works well, data is inserted and table is created. However, since we're using ddb lock store, the following records for delta log files are created in the dynamodb
![image](https://github.com/user-attachments/assets/151e755e-8656-42ec-9efa-b77ac70bb777)

On the next run, table is dropped, s3 delta lake files are deleted, but records in the dynamodb aren't. When dbt-spark executes 'create table' command, Delta tries to write 00000000000000000000.json, but fails, since the corresponding lock record already exists.

It feels like the better way to work with delta lake tables are not drop + create, but create if not exist, truncate + insert. Or merge if that makes more sense.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.