Hello select count(*) not leverage metadata with where condition on partition column this query uses the metadata correctly and answers fast as mentioned in this link https://delta.io/blog/2023-04-19-faster-aggregations-metadata/: select count(*) from table but this query is not: select count(*) from table where partition_col = part_value spark version: 3.3.2 delta version: 2.3

Team, am getting this exception in prod when am creating temp table, usually I will drop this temp table and recreate for every micro batch in spark streaming. some race conditions happening in S3 and Delta?. I just worried to set PERMISSIVE mode, because of silent ignore. Anyone experienced like this error before?. Am using Delta 2.2.0 and Spark 3.3.1 and EMR 6.10.0 versions. Thanks for your help!.

rg.apache.spark.SparkException: Malformed records are detected in record parsing. Parse Mode: FAILFAST. To process malformed records as null result, try setting the option 'mode' as 'PERMISSIVE'.
        at org.apache.spark.sql.errors.QueryExecutionErrors$.malformedRecordsDetectedInRecordParsingError(QueryExecutionErrors.scala:1417) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.catalyst.util.FailureSafeParser.parse(FailureSafeParser.scala:68) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.execution.datasources.json.TextInputJsonDataSource$.$anonfun$readFile$9(JsonDataSource.scala:144) ~[spark-sql_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:486) ~[scala-library-2.12.15.jar:?]
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:492) ~[scala-library-2.12.15.jar:?]
        at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) ~[scala-library-2.12.15.jar:?]
        at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:227) ~[spark-sql_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) ~[scala-library-2.12.15.jar:?]
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source) ~[?:?]
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:35) ~[spark-sql_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.hasNext(Unknown Source) ~[?:?]
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:955) ~[spark-sql_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) ~[scala-library-2.12.15.jar:?]
        at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:168) ~[spark-core_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) ~[spark-core_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) ~[spark-core_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) ~[spark-core_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.scheduler.Task.run(Task.scala:138) ~[spark-core_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548) ~[spark-core_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1516) ~[spark-core_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551) ~[spark-core_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_372]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_372]
        at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_372]
Caused by: org.apache.spark.sql.catalyst.util.BadRecordException: com.fasterxml.jackson.core.io.JsonEOFException: Unexpected end-of-input: expected close marker for Object (start marker at [Source: (byte[])"{"add":{"path":"site_id=76492/part-00190-15100589-5cb2-42a9-ba1d-46ca7f6fc84c.c000.snappy.parquet","partitionValues":{"site_id":"76492"},"size":26061,"modificationTime":1687106147000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":10375265,\"business_entity_id\":1,\"handle\":\"SOME_VALUE\",\"initial_plan_handle\":\"free\",\"current_plan_handle\":\"free\",\"currency_code\":\"USD\",\"created_at\":\"2018-05-18T15:29:17.000Z\",\"activated_at\":\"2018-05-18T15:29:17.000Z\",\"[truncated 1733 bytes]; line: 1, column: 8])
 at [Source: (byte[])"{"add":{"path":"site_id=76492/part-00190-15100589-5cb2-42a9-ba1d-46ca7f6fc84c.c000.snappy.parquet","partitionValues":{"site_id":"76492"},"size":26061,"modificationTime":1687106147000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":10375265,\"business_entity_id\":1,\"handle\":\"SOME_VALUE\",\"initial_plan_handle\":\"free\",\"current_plan_handle\":\"free\",\"currency_code\":\"USD\",\"created_at\":\"2018-05-18T15:29:17.000Z\",\"activated_at\":\"2018-05-18T15:29:17.000Z\",\"[truncated 1733 bytes]; line: 1, column: 2234]
        at org.apache.spark.sql.catalyst.json.JacksonParser.parse(JacksonParser.scala:513) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.execution.datasources.json.TextInputJsonDataSource$.$anonfun$readFile$7(JsonDataSource.scala:140) ~[spark-sql_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.catalyst.util.FailureSafeParser.parse(FailureSafeParser.scala:60) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        ... 22 more
Caused by: com.fasterxml.jackson.core.io.JsonEOFException: Unexpected end-of-input: expected close marker for Object (start marker at [Source: (byte[])"{"add":{"path":"site_id=76492/part-00190-15100589-5cb2-42a9-ba1d-46ca7f6fc84c.c000.snappy.parquet","partitionValues":{"site_id":"76492"},"size":26061,"modificationTime":1687106147000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":10375265,\"business_entity_id\":1,\"handle\":\"SOME_VALUE\",\"initial_plan_handle\":\"free\",\"current_plan_handle\":\"free\",\"currency_code\":\"USD\",\"created_at\":\"2018-05-18T15:29:17.000Z\",\"activated_at\":\"2018-05-18T15:29:17.000Z\",\"[truncated 1733 bytes]; line: 1, column: 8])
 at [Source: (byte[])"{"add":{"path":"site_id=76492/part-00190-15100589-5cb2-42a9-ba1d-46ca7f6fc84c.c000.snappy.parquet","partitionValues":{"site_id":"76492"},"size":26061,"modificationTime":1687106147000,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":10375265,\"business_entity_id\":1,\"handle\":\"SOME_VALUE\",\"initial_plan_handle\":\"free\",\"current_plan_handle\":\"free\",\"currency_code\":\"USD\",\"created_at\":\"2018-05-18T15:29:17.000Z\",\"activated_at\":\"2018-05-18T15:29:17.000Z\",\"[truncated 1733 bytes]; line: 1, column: 2234]
        at com.fasterxml.jackson.core.base.ParserMinimalBase._reportInvalidEOF(ParserMinimalBase.java:682) ~[jackson-core-2.13.4.jar:2.13.4]
        at com.fasterxml.jackson.core.base.ParserBase._handleEOF(ParserBase.java:494) ~[jackson-core-2.13.4.jar:2.13.4]
        at com.fasterxml.jackson.core.base.ParserBase._eofAsNextChar(ParserBase.java:511) ~[jackson-core-2.13.4.jar:2.13.4]
        at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._skipWSOrEnd(UTF8StreamJsonParser.java:3039) ~[jackson-core-2.13.4.jar:2.13.4]
        at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:756) ~[jackson-core-2.13.4.jar:2.13.4]
        at org.apache.spark.sql.catalyst.json.JacksonUtils$.nextUntil(JacksonUtils.scala:30) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at <http://org.apache.spark.sql.catalyst.json.JacksonParser.org|org.apache.spark.sql.catalyst.json.JacksonParser.org>$apache$spark$sql$catalyst$json$JacksonParser$$convertObject(JacksonParser.scala:424) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.catalyst.json.JacksonParser$$anonfun$$nestedInanonfun$makeStructRootConverter$3$1.applyOrElse(JacksonParser.scala:102) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.catalyst.json.JacksonParser$$anonfun$$nestedInanonfun$makeStructRootConverter$3$1.applyOrElse(JacksonParser.scala:101) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.catalyst.json.JacksonParser.parseJsonToken(JacksonParser.scala:377) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.catalyst.json.JacksonParser.$anonfun$makeStructRootConverter$3(JacksonParser.scala:101) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.catalyst.json.JacksonParser.$anonfun$parse$2(JacksonParser.scala:501) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.util.Utils$.tryWithResource(Utils.scala:2860) ~[spark-core_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.catalyst.json.JacksonParser.parse(JacksonParser.scala:496) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.execution.datasources.json.TextInputJsonDataSource$.$anonfun$readFile$7(JsonDataSource.scala:140) ~[spark-sql_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]
        at org.apache.spark.sql.catalyst.util.FailureSafeParser.parse(FailureSafeParser.scala:60) ~[spark-catalyst_2.12-3.3.1-amzn-0.jar:3.3.1-amzn-0]

Hi everyone, I have a use case where I need to regularly delete rows from a partitioned delta table where id comes from a dataframe. I use the following.

val table = DeltaTable.forPath(spark, <path>)

// table schema could be like this
// id: int, ...some other columns, p1: string, p2: string
// here p1 and p2 are partition columns with order p1, p2

val df = <data frame with ids to delete, could be around 10k ids>

table
.merge(
broadcast(df),
col(p1).isin(<some p1 values from which I want to delete ids>) && col("id") === df("id")
)
.whenMatched()
.delete()
.execute()

Now I want to preserve the partitioning structure after the above operation so I set spark.databricks.delta.merge.repartitionBeforeWrite.enabled=true. When I run the above, I notice a significant amount of shuffle. Apparently, spark reads data from all partitions of p1 which match the passed values, then removes the matching rows using a broadcast join and then performs a repartition->partitionBy before writing to preserve partitioning structure (p1->p2). I feel that we can avoid the shuffle here because merge operation is not changing the partitioning structure. Ie. spark can go to each partition, replace the older file with new file from which ids are deleted and create entries for these new files into delta log. Is there any method or config that I can use to avoid the shuffle in the above operation? I currently use delta lake 0.8.0. Thanks in advance!

Hi All, I have a scenario where I need to create a table on top of binary with parquet file format. Can anyone please help me to create delta table for this kind of data fmt. Parquet schema file structure :

Hi,

Is there any way we can update an existing delta table using delta-rs(python) version?

I can only see append and overwrite modes in writing to delta table. but I cant find anything to upsert to existing delta table without having to overwrite the table completely.

I am kind of stuck with update using delta-rs. ANY HELP WOULD BE REALLY APPRECIATED.

Ok, this is mostly a Deltalake and Rust question... I know I can load a delta lake table version using the

version

parameter in

open_table_with_version

, but my question is a little around this, How can I get the changes done in that version? for example, just the inserted rows in that version instead of all the tuples up to that version?

hi i am trying to use generated identity column in my delta table but when i am trying to write to the delta table without the auto generated column getting errror that schema mismatch no of columns of table does not match table columns and if i try passing the identity column with some value its giving error provinding values to generated always column is not supported

Hi, Is there a Synapse Activity that I can use to execute a Select statement against synapse lake database with out using Notebook?

HI all new to delta lake. One thing i've found challenging is determining what features are part of delta lake oss, and what features on proprietary to databricks. Is there an authoritative place to see what settings (tblproperties, spark settings, etc) are used in delta lake and which are only used in databricks?

When I run optimize on a delta lake table, I can specify a target file size like this:

spark.conf.set("spark.databricks.delta.optimize.maxFileSize", 256*1024*1024)

Is there an equivalent setting for the target file size when writing an initial dataset, for example from

df.write

?

Hi team, I'm looking to populate a delta table with timestamps without timezones, and I'll be using AWS Glue and AWS Athena to read this table. Could you please advise on the best approach to achieve this? Thank you!

Hello! My team was facing an interesting data corruption issue with Spark/Delta and we reported it to the Apache Spark maintainers. They created a new setting which is supposed to address this. I wanted to see if anyone had other thoughts on why this issue might be happening and some ideas on other ways that the problem could be addressed? https://issues.apache.org/jira/browse/SPARK-43816

Hellow!!, I'm currently experiencing a datatype mismatch and schema inconsistency issue when querying the same table in both Databricks and AWS Athena. Although the datatype is specified as a timestamp in both cases, when I use the typeof query, I see different output. I'm seeking assistance to understand if there are any missing configurations or settings that need to be adjusted. My goal is to have a timestamp without a time zone. 1st Databricks: 2nd AWS Athena:

I was wondering if this library uses metadata statistics embedded in the Delta log for file filtering? I did a quick read of the code and noticed that file filtering functionality is usually transferred to the PyArrow library. But doesn't this imply that each parquet file is opened to read the parquet statistics?

Hi Team, Please help me to connect Delta-rs and HDFS. I couldn't see any Python documentation.

I recently enabled column mapping on my delta table and I'm having an issue where a bunch of columns are now displaying the physical names instead of the logical names, e.g.

col-04ee4877-ee53-4cb9-b1fb-1a4eb74b508c

. The underlying parquet files still have the right column names (which I think is the whole point of the column mapping right?), but is there a way to force delta table reads to use the logical/parquet file reads?

Hi Delta-Lake Team , Using Spark 3.3 delta-spark== 2.3.0 python 3.9 All set to go ! Unfortunately when running the below pyspark code : from pyspark.sql import SparkSession from delta.pip_utils import configure_spark_with_delta_pip

spark = (
    SparkSession \
    .builder.appName("DeltaLakeFundamentals") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
)

spark = configure_spark_with_delta_pip(spark).getOrCreate()

spark.sql("create table default.np_delta_table (id int , name string) using delta")

I am getting the below error can some one please help .

Traceback (most recent call last):
  File "/home/bchaudhu/delta/delta-create.py", line 13, in <module>
    spark.sql("create table default.np_delta_table (id int , name string) using delta")
  File "/opt/cloudera/parcels/SPARK3-3.3.0.3.3.7180.0-274-1.p0.31212967/lib/spark3/python/lib/pyspark.zip/pyspark/sql/session.py", line 1034, in sql
  File "/opt/cloudera/parcels/SPARK3-3.3.0.3.3.7180.0-274-1.p0.31212967/lib/spark3/python/lib/py4j-0.10.9.5-src.zip/py4j/java_gateway.py", line 1321, in __call__
  File "/opt/cloudera/parcels/SPARK3-3.3.0.3.3.7180.0-274-1.p0.31212967/lib/spark3/python/lib/pyspark.zip/pyspark/sql/utils.py", line 190, in deco
  File "/opt/cloudera/parcels/SPARK3-3.3.0.3.3.7180.0-274-1.p0.31212967/lib/spark3/python/lib/py4j-0.10.9.5-src.zip/py4j/protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o72.sql.
: java.lang.NoSuchMethodError: org.apache.spark.sql.catalyst.catalog.CatalogTable.<init>(Lorg/apache/spark/sql/catalyst/TableIdentifier;Lorg/apache/spark/sql/catalyst/catalog/CatalogTableType;Lorg/apache/spark/sql/catalyst/catalog/CatalogStorageFormat;Lorg/apache/spark/sql/types/StructType;Lscala/Option;Lscala/collection/Seq;Lscala/Option;Ljava/lang/String;JJLjava/lang/String;Lscala/collection/immutable/Map;Lscala/Option;Lscala/Option;Lscala/Option;Lscala/collection/Seq;ZZLscala/collection/immutable/Map;Lscala/Option;)V
        at org.apache.spark.sql.delta.catalog.DeltaCatalog.$anonfun$createDeltaTable$1(DeltaCatalog.scala:134)
        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.catalog.DeltaCatalog.recordFrameProfile(DeltaCatalog.scala:57)
        at <http://org.apache.spark.sql.delta.catalog.DeltaCatalog.org|org.apache.spark.sql.delta.catalog.DeltaCatalog.org>$apache$spark$sql$delta$catalog$DeltaCatalog$$createDeltaTable(DeltaCatalog.scala:86)
        at org.apache.spark.sql.delta.catalog.DeltaCatalog.$anonfun$createTable$1(DeltaCatalog.scala:271)
        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.catalog.DeltaCatalog.recordFrameProfile(DeltaCatalog.scala:57)
        at org.apache.spark.sql.delta.catalog.DeltaCatalog.createTable(DeltaCatalog.scala:263)
        at org.apache.spark.sql.execution.datasources.v2.CreateTableExec.run(CreateTableExec.scala:45)
        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$1.$anonfun$applyOrElse$1(QueryExecution.scala:99)
        at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:109)
        at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:169)
        at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:95)
        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:779)
        at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
        at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:99)
        at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:94)
        at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:584)
        at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:176)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:584)
        at <http://org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org|org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org>$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:30)
        at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
        at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:560)
        at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:94)
        at org.apache.spark.sql.execution.QueryExecution.commandExecuted$lzycompute(QueryExecution.scala:81)
        at org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:79)
        at org.apache.spark.sql.Dataset.<init>(Dataset.scala:220)
        at org.apache.spark.sql.Dataset$.$anonfun$ofRows$2(Dataset.scala:100)
        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:779)
        at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:97)
        at org.apache.spark.sql.SparkSession.$anonfun$sql$1(SparkSession.scala:622)
        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:779)
        at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:617)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
        at py4j.Gateway.invoke(Gateway.java:282)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
        at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
        at java.lang.Thread.run(Thread.java:748)

Where I am going wrong ? Can some one please guide .

[ Delta-RS Question] When I load a delta table with the without_files=True parameter, can it be effective when executing commands like deltable.vacuum and dt.optimize.compact()? example: dt = DeltaTable(storage_options=storage_options, table_uri=f{table_minio_url}, without_files=True) dt.vacuum() dt.optimize.compact()

Can some one please help me on this ? https://delta-users.slack.com/archives/CJ70UCSHM/p1687518830259069

Just wanted to follow up on this! Does anyone have any ideas on it?

Hi. Does anyone knows if it is possible to use deltalake-rs for streaming with datafusion in Rust? I can’t seem to find anything on this

Hi everyone. I'm using delta on a cluster with huge amount of data every day so i should delete files older than 5 days and i use delta VACUUM. but i have encountered into a big problem. when i use VACUUM delta must delete invalidated files but it also deletes some validated files and it's really strange to me. it actually deletes files older than 5 days and also deletes files from current day which its data is completing. I also have set the logRetentionDuration config to delete old logs. I would really appreciate a solution. delta_table = DeltaTable.forPath(session , path) delta_table.logRetentionDuration = "interval 30 days" delta_table.deletes( col("date") < "specified_day" ) session.sql("vacuum delta.path RETAIN 0 HOURS") session.sql("vacuum delta.path RETAIN 0 HOURS")

I'm trying to upsert data from a kafka stream into a large-ish delta lake dataset (about 35B rows) using spark structured streaming. Upserted data is relatively small per batch, maybe 1-2m records. The data is partitioned, I'm using a partition field condition in the merge query, and it only affects modest subset of the partitions, but I still see pyspark/delta scan the entire dataset. The table is also zordered by the columns I'm matching on for the upsert. Does delta lake oss upsert use partitions and zordered columns for data skipping? Is it expected that delta needs to scan the entire table to upsert? I'd expect partition pruning at least to make this more efficient but I'm not seeing it. Edit: I've removed streaming from the equation and am just trying to get a series of parquet files uploaded efficiently and seeing the same behavior as above

How do you get

delta-rs

to automatically write checkpoint files? I’ve got around 300 json files in my _delta_log but no checkpoints. Is it something you have to do manually? Code here for context https://github.com/helium/helium-data/blob/main/protobuf-delta-lake-sink/src/main.rs#L304

Hi team, Just I am trying to understand at which level Delta file format is operating... Does Delta file format act like a server (a process keeps running and responds to requests) ? Analogous to Hadoop file system, where the base file system is Unix File System and on top of it, HDFS operates where name node manages the HDFS files. But if this is to be true, I don't see any server process described in Delta file format. So, what is responsible for its features (ACID, upsert, delete, data governance etc..) ? To add on, it is told that many other tools can interact with Delta file format. For example: DBT (a SQL based transformation tool) can read/write data. If this is the case, which process is responsible to provide the aforementioned features? Also it is mentioned that Delta format supports only tables. If yes, is it a RDBMS product? For HDFS, it very clear that it operates on top of host OS file system and different processes (name node, data'node etc) are available to interact with it. Similarly I don't get any clarity about the Delta format. Any help will be much appreciated. Thanks

In delta-rs, when trying to write a checkpoint, I’m getting

Error: Failed to convert into Arrow schema: Json error: Binary is not supported by JSON

Any idea how to fix?

Hi Team , will i get duplicate records if i re-ran partial failed merge command (some records have been updated or inserted) on the same target delta table but source delta lake gets updated each time. Assume the following scenario : Note: Each record contains the primary key based on which merge is performed. In Run-1, we have source S1 with 10 records ( 5- update , 5- insert ) and Target with 100 records after merge, target delta lake will contain 105 records. In Run-2, we have source S2 with 5 records ( 2-update , 3- insert) and Target with 105 records and merge have failed after writing partially ( 2 records have been inserted and 1 record have been updated ). As above Run-2 have failed, we re-ran the job again. But this time S2 (source delta table got updated) will contain 7 records ( 3-update,4-insert) which includes the previous 5 records (2- update , 3- insert) + current 2 records ( 1 -update, 1- insert ) . So if i do merge now on the target delta table, will it insert the already inserted records and create duplicates ? Any help on this issue ?